Basic initialization

from smartmemory import SmartMemory

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

memory = SmartMemory()

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

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

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]]

# 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
)

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

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)

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

# 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"
)

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

# 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")

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

# 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)

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

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

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

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

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

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

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

from smartmemory import MemoryItem

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

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

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

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

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

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

# 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.

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

# 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")

def clear(self) -> None

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

def resolve_external(self, node: MemoryItem) -> Optional[List[Any]]

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}")

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 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}
})

def run_evolution_cycle(self) -> None

# Trigger evolution cycle
memory.run_evolution_cycle()

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

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)

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)

def run_clustering(self) -> dict

# 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")

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

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

def summary(self) -> Dict

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

def orphaned_notes(self) -> List

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

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

# 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

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

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'
)

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']}")

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

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

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

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

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

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)

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

# 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")

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

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

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

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