Basic initialization

The SmartMemory class is the main entry point for all memory operations. It provides a unified interface for storing, retrieving, and managing different types of memories with intelligent processing capabilities.

Class: SmartMemory

from smartmemory import SmartMemory

memory = SmartMemory(
    config_path: Optional[str] = None,
    config: Optional[Dict] = None,
    **kwargs
)

Constructor Parameters

Example

memory = SmartMemory()

# With custom configuration
memory = SmartMemory(config_path="config.json")

# With inline configuration
memory = SmartMemory(config={
    "graph": {"backend": "FalkorDBBackend"}
})

Core Methods

ingest()

Ingest content with full 11-stage intelligent processing pipeline:

classify → coreference → simplify → entity_ruler → llm_extract →
ontology_constrain → store → link → enrich → ground → evolve

def ingest(
    self,
    item: Union[str, Dict, MemoryItem],
    context: Optional[Dict] = None,
    adapter_name: Optional[str] = None,
    converter_name: Optional[str] = None,
    extractor_name: Optional[str] = None,
    enricher_names: Optional[List[str]] = None,
    conversation_context: Optional[Union[ConversationContext, Dict[str, Any]]] = None,
    pipeline_config: Optional[PipelineConfigBundle] = None,
    sync: Optional[bool] = None,
    auto_challenge: Optional[bool] = None,
    extract_decisions: bool = False,
    extract_reasoning: bool = False,
    **kwargs
) -> Union[str, Dict[str, Any]]

Parameters

Returns

• str - The item_id (when sync=True)
• Dict[str, Any] - {"item_id": str, "queued": bool} (when sync=False)

Examples

# Simple ingestion (full pipeline)
item_id = memory.ingest("I learned Python programming in 2020")

# With user isolation
item_id = memory.ingest(
    "Meeting with John about project timeline",
    user_id="alice"
)

# Async ingestion (queue for background)
result = memory.ingest("Large document...", sync=False)
print(f"Queued: {result['item_id']}, queued: {result['queued']}")

# With conversation context
from smartmemory.conversation.context import ConversationContext
conv_ctx = ConversationContext(conversation_id="conv_123")
item_id = memory.ingest(
    "User prefers Python for data science",
    conversation_context=conv_ctx
)

add()

Simple storage without the full ingestion pipeline. Use for internal operations, derived items, or when pipeline is not needed.

def add(
    self,
    item: Union[str, Dict, MemoryItem],
    **kwargs
) -> str  # item_id

Parameters

Returns

str - The item_id of the stored memory item

Examples

from smartmemory import MemoryItem

# Simple storage (no extraction/linking/evolution)
item = MemoryItem(content="Quick note", memory_type="semantic")
item_id = memory.add(item)

# With metadata
item = MemoryItem(
    content="Derived fact",
    memory_type="semantic",
    metadata={"source": "enrichment"}
)
item_id = memory.add(item)

When to use which:

• ingest() - User input, external data, anything needing full processing
• add() - Internal operations, derived items, evolution/enrichment output

ingest_structured()

Ingest structured data via a registered handler, bypassing the NLP pipeline. The handler determines the ingestion strategy (FULL/INDEXED/APPEND), validates input, and declares edges and indexes.

def ingest_structured(
    self,
    data: dict,
    schema: str,
    **kwargs
) -> str  # item_id

Parameters

Returns

str - The item_id of the stored item

Available Schemas

Examples

# Store a decision
item_id = memory.ingest_structured(
    data={
        "title": "Use FalkorDB for graph storage",
        "status": "accepted",
        "rationale": "Native graph + vector support",
        "alternatives": ["Neo4j", "ArangoDB"]
    },
    schema="decision"
)

# Record a tool call
item_id = memory.ingest_structured(
    data={
        "tool": "memory_search",
        "input": {"query": "Python frameworks"},
        "output": {"results": 5}
    },
    schema="tool_call"
)

# Store a plan via PlanManager
item_id = memory.ingest_structured(
    data={
        "title": "Migration Plan",
        "phases": [{"name": "Phase 1", "tasks": ["Setup", "Migrate"]}]
    },
    schema="plan"
)

ingest_code()

Index a code repository, creating searchable graph nodes for classes, functions, routes, and tests.

def ingest_code(
    self,
    directory: str,
    repo: str,
    commit_hash: str | None = None,
    exclude_dirs: set[str] | None = None,
    languages: list[str] | None = None,
) -> IndexResult

Parameters

Returns

IndexResult with the following fields:

Example

# Index a Python repository
result = memory.ingest_code(
    directory="/path/to/my-project",
    repo="my-project",
    exclude_dirs={"node_modules", ".venv", "__pycache__"}
)

print(f"Indexed {result.entities_created} entities")
print(f"Created {result.edges_created} edges")
print(f"Generated {result.embeddings_generated} embeddings")
print(f"Completed in {result.elapsed_seconds:.1f}s")

search()

Search for memories using various strategies.

def search(
    self,
    query: str,
    top_k: int = 5,
    memory_type: Optional[str] = None,
    expertise: bool = False
) -> Union[List[MemoryItem], Dict[str, List]]

Cross-encoder evidence on search results

GET /memory/search and POST /memory/search return these fields on each item, including expertise buckets and citation modes. Python SDK MemoryItem preserves them; the JavaScript SDK passes them through. They expose existing inference only and do not apply an admission threshold or change ranking.

null never means zero or implicitly relevant. An unscored tail can move during later temporal/strength ordering, so check each returned item's status. Items inserted after reranking are also unverified for that query. Uninstrumented/legacy paths report not_reranked with null diagnostics. Missing fields from older servers also provide no evidence of relevance.

The ordinary search call currently uses effective defaults of 25 candidates and 512 document characters; custom RerankerConfig.max_candidates and max_doc_chars remain unforwarded on this path because activating them would change behavior. Counts describe the original pool, not the returned page. Unknown model revision or activation is insufficient evidence for comparing thresholds across model deployments. No relevance gate or threshold is shipped by this exposure change.

Parameters

Note: User/tenant filtering is handled automatically by ScopeProvider. No manual user_id parameter needed.

Returns

• List[MemoryItem] for standard search
• dict[str, list] when expertise=True, keyed by decision, constraint, learned, opinion, reasoning, and observation

Examples

# Basic search
results = memory.search("Python programming")

# Type-specific search
semantic_results = memory.search("France", memory_type="semantic")
episodic_results = memory.search("meeting", memory_type="episodic")

# Limited results
top_results = memory.search("AI", top_k=3)

# Expertise search returns buckets by expertise type
expertise_results = memory.search("storage decisions", expertise=True)
for decision in expertise_results["decision"]:
    print(decision.decision_id)

Expertise Capture

The expertise capture methods return typed objects, not string IDs.

def add_decision(self, content: str, **kwargs) -> "Decision"
def add_constraint(self, content: str, **kwargs) -> "Constraint"
def add_learning(self, content: str, **kwargs) -> "Learned"

• add_decision(...) returns a Decision with content, rejected_alternatives, rationale, decision_type, status, decision_id, confidence, constraints, and domain.
• add_constraint(...) returns a Constraint with .constraint_id.
• add_learning(...) returns a Learned with .learned_id.

get()

Retrieve a specific memory item by ID.

def get(self, item_id: str, **kwargs) -> Optional[MemoryItem]

Parameters

Returns

Optional[MemoryItem] - The memory item if found, None otherwise

Example

# Get specific memory
memory_item = memory.get("memory_id_123")
if memory_item:
    print(f"Content: {memory_item.content}")

update()

Update an existing memory item.

def update(self, item: Union[MemoryItem, Dict]) -> str  # item_id

Parameters

Returns

str - The item_id of the updated item

Example

# Update existing memory
memory_item = memory.get("memory_id_123")
memory_item.content = "Updated content"
updated_item = memory.update(memory_item)

delete()

Delete a memory item.

def delete(self, item_id: str, **kwargs) -> bool

Parameters

Returns

bool - True if deleted successfully, False otherwise

Example

# Delete memory
success = memory.delete("memory_id_123")
print(f"Deleted: {success}")

MemoryItem Fields

The MemoryItem class represents a single memory record. All fields are available on items returned by get(), search(), and other retrieval methods.

from smartmemory import MemoryItem

Origin Taxonomy

The origin field tracks how a memory was created. Prefixes are organized into 4 visibility tiers in smartmemory/origin_policy.py. Common patterns:

Relationship Methods

link()

Create explicit relationships between memories.

def link(
    self,
    source_id: str,
    target_id: str,
    link_type: Union[str, LinkType] = "RELATED"
) -> str  # "Linked <src> to <tgt> as <type>"

Parameters

Returns

str - Confirmation string

Example

# Create relationship
memory.link("python_memory", "project_memory", "USED_FOR")

get_links()

Get all links for a memory item.

def get_links(
    self,
    item_id: str,
    memory_type: str = "semantic"
) -> List[str]  # List of link/triple strings

Parameters

Returns

List[str] - List of link/triple strings for all edges involving item_id

Example

# Get all links
links = memory.get_links("memory_id_123")
for link in links:
    print(link)  # Prints link string representation

get_neighbors()

Get neighboring memories through relationships.

def get_neighbors(self, item_id: str, direction: str = "both") -> List[MemoryItem]

Parameters

Returns

List[MemoryItem] - List of neighboring memory items

Example

# Get related memories
neighbors = memory.get_neighbors("memory_id_123")
print(f"Found {len(neighbors)} related memories")

Background Processing

SmartMemory supports synchronous and asynchronous ingestion via the ingest() method.

# Synchronous (process now via full pipeline)
memory.ingest("Process immediately", sync=True)

# Asynchronous (quick persist + enqueue for background workers)
result = memory.ingest("Process in background", sync=False)
print(result)  # {"item_id": "...", "queued": True}

# Note: a separate worker service must consume the background queue.

Utility Methods

ground()

Ground a memory item to an external source for provenance and validation.

def ground(
    self,
    item_id: str,
    source_url: str,
    validation: Optional[Dict] = None
) -> None

Parameters

Description

Grounding establishes provenance by linking a memory item to its external source. This is crucial for:

• Fact verification: Link memories to authoritative sources
• Audit trails: Track the origin of information
• Quality assurance: Associate validation metadata
• Transparency: Enable source traceability for AI decisions

Example

# Ground a memory to its source
memory.ground(
    item_id="fact_123",
    source_url="https://example.com/article",
    validation={"confidence": 0.95, "verified_at": "2024-01-01"}
)

# Ground with minimal information
memory.ground("memory_456", "https://source.com/data")

clear()

Clear all memory data.

def clear(self) -> None

Example

# Clear all memories (use with caution!)
memory.clear()

Error Handling

SmartMemory does not expose a broad public exception hierarchy. Core operations (add, ingest, search) surface standard Python exceptions, backend/connection failures propagate as the underlying driver errors (e.g. redis/falkordb exceptions), and invalid arguments raise the usual ValueError / TypeError. Wrap calls in a generic handler:

try:
    result = memory.add("content")
    results = memory.search("query")
except Exception as e:
    # Backend/connection failures and invalid input both surface here
    print(f"Memory operation failed: {e}")

Managed-type errors

The only first-class exceptions SmartMemory raises are from the managed-type layer (schema-validated structured types). They live in smartmemory.managed.exceptions:

from smartmemory.managed.exceptions import ManagedTypeError, QualityGateRejected

try:
    memory.ingest_structured(payload, schema="tool_call")
except QualityGateRejected as e:   # subclass of ManagedTypeError
    print(f"Rejected by quality gate: {e}")
except ManagedTypeError as e:
    print(f"Managed-type validation failed: {e}")

Configuration Properties

Access Configuration

# Configuration is set during initialization
memory = SmartMemory(config_path="config.json")

# Or with inline configuration
memory = SmartMemory(config={
    "graph": {"backend": "FalkorDBBackend"},
    "similarity": {"semantic_weight": 0.5}
})

Note: Configuration is set at initialization time. Runtime configuration updates are not supported.

Performance Tips

1. Batch operations when possible:

   items = ["item1", "item2", "item3"]
   for item in items:
       memory.ingest(item, sync=False)  # Enqueue for background workers

2. Cache frequently accessed memories:

   # Store frequently used memory IDs
   important_memories = ["id1", "id2", "id3"]
   cached_memories = {id: memory.get(id) for id in important_memories}

3. Use specific memory types for better search performance:

   # More efficient than general search
   results = memory.search("query", memory_type="semantic")

Additional Methods

Evolution and Memory Management

run_evolution_cycle()

Trigger a memory evolution cycle for consolidation and optimization.

def run_evolution_cycle(self) -> None

Example:

# Trigger evolution cycle
memory.run_evolution_cycle()

challenge()

Challenge an assertion against existing knowledge to detect contradictions using a multi-method detection cascade.

def challenge(
    self,
    assertion: str,
    memory_type: str = "semantic",
    use_llm: bool = True
) -> ChallengeResult

Parameters:

• assertion: The new fact/assertion to challenge
• memory_type: Type of memory to search (default: "semantic")
• use_llm: Whether to use LLM for deep contradiction analysis

Returns: ChallengeResult with detected conflicts

Detection Cascade:

1. LLM (if use_llm=True) - Most accurate, analyzes nuance and context
2. Graph - Detects functional property conflicts (capital, CEO, born in)
3. Embedding - High semantic similarity + opposite polarity
4. Heuristic - Pattern matching fallback

Example:

result = memory.challenge("Paris is the capital of Germany")

if result.has_conflicts:
    for conflict in result.conflicts:
        print(f"Contradicts: {conflict.existing_fact}")
        print(f"Method: {conflict.explanation}")  # Shows [LLM], [Graph], etc.
        print(f"Confidence: {conflict.confidence}")
        
# Fast challenge (skip LLM)
result = memory.challenge("Some fact", use_llm=False)

Using AssertionChallenger directly:

from smartmemory.reasoning import AssertionChallenger, ResolutionStrategy

challenger = AssertionChallenger(
    memory,
    use_llm=True,      # LLM detection
    use_graph=True,    # Graph structure analysis
    use_embedding=True, # Semantic + polarity
    use_heuristic=True  # Pattern matching
)

result = challenger.challenge("Paris is the capital of Germany")

# Auto-resolve conflicts (tries Wikipedia, LLM, grounding, recency)
for conflict in result.conflicts:
    resolution = challenger.resolve_conflict(conflict, auto_resolve=True)
    
    if resolution["auto_resolved"]:
        print(f"Auto-resolved via {resolution['method']}")
        print(f"Evidence: {resolution['evidence']}")
    else:
        # Fall back to manual strategy
        challenger.resolve_conflict(conflict, strategy=ResolutionStrategy.DEFER)

Auto-Resolution Methods:

1. Wikipedia - Looks up entities, checks which fact aligns (0.85 confidence)
2. LLM Reasoning - GPT fact-checks with reasoning (0.7+ confidence required)
3. Grounding - Checks existing provenance/trusted sources (0.75 confidence)
4. Recency - For temporal conflicts, prefers recent info (0.65 confidence)

run_clustering()

Run entity clustering to deduplicate entities across the graph.

def run_clustering(self) -> dict

Returns: Dictionary with clustering statistics

Example:

# Run clustering (SemHash + embedding + LLM)
stats = memory.run_clustering()
print(f"Merged {stats.get('merged_count', 0)} duplicate entities")
print(f"Found {stats.get('clusters_found', 0)} clusters")

Clustering Pipeline:

1. SemHash pre-deduplication - Fast deterministic dedup (0.95 threshold)
2. KMeans embedding clustering - Group similar entities (~128 per cluster)
3. LLM semantic clustering - Find aliases (Joe ↔ Joseph, ML ↔ machine learning)
4. Graph node merging - Rewire edges, merge properties

commit_working_to_episodic() / commit_working_to_procedural() (REMOVED)

Removed in CORE-MEMORY-DYNAMICS-1 M1 (2026-04). The retroactive-promotion pattern was replaced by the ConsolidationRouter pipeline stage, which routes items to episodic/procedural/semantic/drop at ingest time based on content signals. Items awaiting a routing decision live under memory_type="pending" (renamed from "working") and the router's decision is observable on PipelineState.router_decision.

Migration: instead of explicit commit_working_to_episodic(), store items with memory_type="pending" (or let the pipeline's auto-classification pick a type) and the router will handle consolidation.

Graph Operations

add_edge()

Add an edge between two memory nodes.

def add_edge(
    self,
    source_id: str,
    target_id: str,
    relation_type: str,
    properties: Optional[Dict] = None
) -> Any

create_or_merge_node()

Create or merge a node with properties.

def create_or_merge_node(
    self,
    item_id: str,
    properties: Dict,
    memory_type: Optional[str] = None
) -> str

Monitoring and Analytics

summary()

Get summary statistics about memory contents.

def summary(self) -> Dict

Example:

stats = memory.summary()
print(f"Total items: {stats.get('total_count', 0)}")

orphaned_notes()

Find orphaned memory items (no connections).

def orphaned_notes(self) -> List

prune()

Prune old or unused memories.

def prune(self, strategy: str = "old", days: int = 365, **kwargs) -> Any

Temporal Queries

time_travel()

Context manager for temporal queries at a specific point in time.

def time_travel(self, to: str) -> ContextManager

Example:

# Query memories as they existed on a specific date
with memory.time_travel("2024-09-01"):
    results = memory.search("Python")
    # Results reflect state on Sept 1st, 2024

Version Tracking

SmartMemory provides bi-temporal versioning via version_tracker:

version_tracker.get_versions()

Get all versions of a memory item.

versions = memory.version_tracker.get_versions(item_id)
for v in versions:
    print(f"Version {v.version_number}: {v.content[:50]}...")

version_tracker.get_version_at_time()

Get the version that was current at a specific time.

from datetime import datetime

version = memory.version_tracker.get_version_at_time(
    item_id="mem_123",
    time=datetime(2024, 1, 15),
    time_type='transaction'  # or 'valid'
)

version_tracker.compare_versions()

Compare two versions of an item.

diff = memory.version_tracker.compare_versions(
    item_id="mem_123",
    version1=1,
    version2=2
)
print(f"Content changed: {diff['content_changed']}")
print(f"Metadata changes: {diff['metadata_changes']}")

Archive Operations

archive_put()

Archive a conversation artifact durably.

def archive_put(
    self,
    conversation_id: str,
    payload: Union[bytes, Dict[str, Any]],
    metadata: Dict[str, Any]
) -> Dict[str, str]

Returns: Dictionary with archive_uri and content_hash

archive_get()

Retrieve an archived artifact by URI.

def archive_get(self, archive_uri: str) -> Union[bytes, Dict[str, Any]]

Tags and Metadata

add_tags()

Add tags to a memory item.

def add_tags(self, item_id: str, tags: List[str]) -> bool

Example:

memory.add_tags("memory_123", ["important", "reference", "python"])

update_properties()

Update memory node properties with merge/replace semantics.

def update_properties(
    self,
    item_id: str,
    properties: Dict,
    write_mode: Optional[str] = None
) -> Any

Thread Safety

SmartMemory is thread-safe for read operations but requires coordination for write operations:

import threading

# Safe for concurrent reads
def read_worker():
    results = memory.search("query")
    return results

# Coordinate writes
write_lock = threading.Lock()

def write_worker(content):
    with write_lock:
        memory.add(content)

Graph Integrity Methods

Added in v0.3.8

delete_run()

Delete all entities created by a specific pipeline run.

def delete_run(self, run_id: str) -> int

Parameters

Returns

int - Number of entities deleted

Example

# Ingest returns run_id for tracking
result = memory.ingest("Some content about Python and AI")
run_id = result.get("run_id")

# Later, clean up all entities from that run
deleted_count = memory.delete_run(run_id)
print(f"Deleted {deleted_count} entities")

Use cases:

• Clean up failed/partial pipeline runs
• Re-run extraction on same content without duplicates
• Undo ingestion for rollback scenarios

rename_entity_type()

Rename an entity type across all matching entities in the graph.

def rename_entity_type(self, old_type: str, new_type: str) -> int

Parameters

Returns

int - Number of entities updated

Example

# Rename "Framework" entities to "Library"
updated = memory.rename_entity_type("Framework", "Library")
print(f"Updated {updated} entities")

Use cases:

• Ontology evolution without re-extraction
• Fix inconsistent entity type names
• Merge similar concepts under canonical names

merge_entity_types()

Merge multiple entity types into a single target type.

def merge_entity_types(self, source_types: List[str], target_type: str) -> int

Parameters

Returns

int - Total number of entities updated

Example

# Consolidate fragmented types
updated = memory.merge_entity_types(
    ["JS_Framework", "Python_Framework", "Web_Framework"],
    "Library"
)
print(f"Merged {updated} entities into 'Library' type")

Use cases:

• Consolidate fragmented entity types from different extraction runs
• Simplify ontology by collapsing related types
• Prepare for ontology template migration

Next Steps

• Memory Types API - Specific memory type interfaces
• Tools API - MCP tools and integrations

Creation-time search windows

SmartMemory.search(query, since=None, until=None) and SmartMemory.search_by_metadata(filters, top_k=5, since=None, until=None) accept ISO-8601 strings or Python datetimes. They filter storage created_at using the half-open interval [since, until). Naive inputs mean UTC; offsets are converted to UTC. An invalid timestamp or since > until raises ValueError.

Metadata filters retain equality and also accept the closed operators gte, gt, lte, lt, for example {"created_at": {"gte": "2026-09-01", "lt": "2026-09-03"}}. Operands are bound query parameters; unknown operators raise.

POST /memory/search accepts optional since and until strings. The existing GET /memory/by-metadata also accepts them. Invalid windows return HTTP 400 with the offending value. Missing bounds are omitted. Responses carry coverage when creation-time accuracy is uncertain, including an explicitly unknown created_at_backfilled_boundary: null. Legacy rows may carry their first-update time. Core list results expose this via results.diagnostics.coverage.

The Python HTTP SDK accepts since and until on both search() and search_by_metadata(); client.last_search_coverage preserves the search envelope's caveat even for empty results.

as_of_date selects the historical corpus; creation bounds filter within it. GET /memory/temporal/search/during answers a different question: matching versions by version timestamp. If version tracking is unavailable it logs a warning and marks current-item rows range_applied: false.

Choosing a multi-hop planner

hop_strategy accepts consensus, relevance, or semantic on POST /memory/search, Python SDK search(), and MCP memory_search. JS SDK uses memory.search(query, { multiHop: true, hopStrategy: 'relevance' }). The Python equivalent is client.search(query, multi_hop=True, hop_strategy="relevance").

• consensus follows entities several top results agree on and may stop before a one-off bridge when distractors lower its frequency score.
• relevance follows entities from the best results, including one-off bridges.
• semantic asks an LLM to plan follow-up searches and incurs LLM calls.

Omitted/null leaves core resolution unchanged: semantic when semantic hops/config is enabled, otherwise consensus. The measured fixture spike kept consensus: bridge F1 improved by 22.92 points with relevance, but median hops increased from 1 to 3, exceeding the pre-registered one-extra-hop bound. This is a fixture result, not a production or LoCoMo quality claim.

Invalid strategies return HTTP 400 naming all three values. A supplied strategy with multi_hop=false is labelled in response inert_parameters; Python SDK retains this in last_inert_parameters, and MCP echoes it.

CLI: sm search "which team owns checkout?" --multi-hop --max-hops 3 --hop-strategy relevance. --max-hops accepts 1–10; --hop-strategy without --multi-hop is rejected. Daemon, local fallback and remote wrapper paths carry the flags.