Ingestion Flow

SmartMemory's ingestion pipeline transforms raw input into enriched, interconnected memories through a sophisticated multi-stage process.

Overview

The ingestion flow consists of 11 stages that process memories from initial input to final storage:

1. classify - Determine memory type (semantic, episodic, procedural, pending, zettel)
2. coreference - Resolve pronouns and entity mentions
3. simplify - Optional sentence simplification for cleaner extraction
4. entity_ruler - Pattern-based entity tagging (public knowledge patterns + workspace patterns)
5. llm_extract - LLM entity & relation extraction (SpacyExtractor as deterministic fallback)
6. ontology_constrain - Constrain extracted types against the active ontology
7. store - Persist memory node + entity nodes (and embeddings) in FalkorDB
8. link - Connect the new item to related existing memories
9. enrich - Optional enrichers (basic, sentiment, topic, temporal, link expansion, skills/tools)
10. ground - Link entities to Wikidata/Wikipedia via GROUNDED_IN edges
11. evolve - Run evolvers (episodic→semantic, episodic→zettel, decay, opinion/observation synthesis, …)

Entity clustering and bi-temporal version records run as separate post-pipeline operations rather than as in-pipeline stages — see Hybrid Storage and memory.run_clustering().

Stage Details

1. Input Processing

# Raw input can be various formats
memory.ingest("I learned Python programming in 2020")
memory.ingest({
    "content": "Meeting notes from today",
    "metadata": {"participants": ["Alice", "Bob"]},
    "memory_type": "episodic"
})

Note: Use ingest() for full pipeline processing. Use add() for simple storage without extraction/linking/evolution. Use ingest_structured() for schema-mapped data that should bypass the NLP pipeline entirely — see Structured Ingestion below.

Processing:

• Content validation and sanitization
• Metadata extraction and normalization
• Format standardization

2. Content Analysis

SmartMemory supports multiple extractors for entity and relationship extraction, each with different capabilities and performance characteristics.

Extraction Types

Entity Extraction:

• People, places, organizations
• Dates, times, durations
• Technical concepts and terms
• Actions and events

Triple/Relationship Extraction:

• Subject-predicate-object triples
• Temporal relationships (BEFORE, AFTER, DURING)
• Causal connections (CAUSES, RESULTS_IN)
• Hierarchical structures (PART_OF, CONTAINS)
• Conversational flow (RESPONDS_TO, FOLLOWS)

Ontology Integration (Optional):

• Formal ontology creation and management
• Entity type hierarchies
• Relationship type definitions
• Knowledge graph schema enforcement

Note: Triple extraction and ontology management are separate concerns. Triple extraction can be enabled without full ontology management for lightweight relationship modeling.

Available Extractors

The shipped plugins live in smartmemory/plugins/extractors/. The llm_extract pipeline stage routes through whichever extractor is configured; SpacyExtractor is used as a deterministic fallback when LLM extraction is unavailable.

1. LLMExtractor (llm) — default LLM-based extractor with structured triple output (entities + relationships).

2. LLMSingleExtractor (llm_single) — single-pass LLM variant; lower latency, used by the lite/CC profile and as a CC default. GroqExtractor is a Groq-API subclass selected automatically when GROQ_API_KEY is set.

3. ConversationAwareLLMExtractor — LLMExtractor subclass that adds short-term conversation context for chat-style ingestion.

4. ReasoningExtractor — extracts chain-of-thought / reasoning traces from text containing explicit reasoning (used for the reasoning extended memory type).

5. DecisionExtractor — extracts structured decision claims (used for the decision extended memory type).

6. SpacyExtractor (spacy) — local spaCy-based NER fallback for entity extraction without LLM access.

Provider routing

PipelineModelRouter maps stages (e.g. llm_extract) to model identifiers and providers; presets include cost_optimized(), quality_optimized(), and balanced(). Precedence: explicit model= > router > config > get_default_model(). See smartmemory/pipeline/model_router.py.

When no LLM is available, PipelineConfig.lite() disables LLM extraction and the pipeline falls back to SpacyExtractor (DEGRADE-1d auto-detects this from OPENAI_API_KEY / GROQ_API_KEY env vars).

Large Text Handling

Texts > 8000 characters are automatically chunked:

• Split by sentence boundaries
• Processed in parallel (ThreadPoolExecutor)
• Results aggregated with entity deduplication

3. Memory Classification

Automatic Classification:

• Temporal markers → Episodic
• Factual statements → Semantic
• Process descriptions → Procedural
• Context-dependent → Pending

Manual Override:

memory.add(content, memory_type="semantic", force=True)

4. Enrichment

Semantic Enhancement:

• Concept expansion and synonyms
• Related topic identification
• Contextual information addition
• Knowledge graph integration

Temporal Processing:

• Time normalization
• Event sequencing
• Duration calculation
• Temporal relationship mapping

5. Linking

Similarity-Based Linking:

• Semantic similarity using embeddings
• Temporal proximity for episodic memories
• Conceptual overlap detection
• Entity co-occurrence analysis

Explicit Relationship Creation:

• Causal relationships
• Part-whole relationships
• Temporal sequences
• Conceptual hierarchies

6. Vector Storage

FalkorDB HNSW Index:

• Native vector indexing with vecf32 type
• Configurable HNSW parameters (M, efConstruction, efRuntime)
• Cosine similarity search
• Automatic tenant isolation via ScopeProvider

7. Enrichment

Wikipedia Integration:

• Lookup entities in Wikipedia
• Add summaries, categories, URLs
• Create enrichment metadata

8. Grounding

Provenance Linking:

• Create Wikipedia nodes (shared globally)
• Create GROUNDED_IN edges from entities to Wikipedia
• Track source attribution

9. Evolution

Memory Promotion:

• Pending → Episodic (threshold: 3+ items)
• Pending → Procedural (threshold: 5+ items)
• Episodic → Semantic (stable facts)
• Episodic decay and archival

10. Clustering

Entity Deduplication:

• SemHash pre-deduplication (0.95 threshold)
• KMeans embedding clustering (~128 items per cluster)
• LLM semantic clustering (Joe ↔ Joseph)
• Graph node merging

11. Versioning

Bi-Temporal Tracking:

• Valid time (when fact was true)
• Transaction time (when recorded)
• Version history with HAS_VERSION edges
• Time-travel queries

Configuration

The pipeline is controlled at two levels: a profile set at construction time that applies to every ingest, and per-call selectors that override the pipeline for a single ingest().

Construction-time profile and stage toggles:

from smartmemory import SmartMemory
from smartmemory.pipeline.config import PipelineConfig

memory = SmartMemory(
    pipeline_profile=PipelineConfig.lite(),  # preset: disable network/LLM-bound stages
    compaction="standard",                   # strip extraction intermediates after store
    enable_ontology=False,                   # omit the ontology_constrain stage entirely
    observability=False,                     # disable metrics/event emission
)

Per-call component selection — swap or scope individual stages for one ingest:

item_id = memory.ingest(
    "Quarterly planning notes...",
    extractor_name="spacy",              # choose the extraction stage
    enricher_names=["sentiment", "topic"],  # run exactly these enrichers
    extract_decisions=True,              # toggle decision extraction
    sync=True,                           # run inline vs. queue for background
)

For the full ingest() signature and every selector, see the SmartMemory API reference.

Performance Characteristics

• Fast Ingestion: Immediate storage with background enrichment
• Scalable Processing: Parallel pipeline stages
• Fault Tolerance: Graceful degradation and retry mechanisms
• Memory Efficiency: Streaming processing for large inputs

Monitoring and Debugging

# Enable detailed logging
memory.set_log_level("DEBUG")

# Access ingestion metrics
stats = memory.get_ingestion_stats()
print(f"Processed: {stats.total_memories}")
print(f"Average processing time: {stats.avg_processing_time}ms")

The ingestion flow is designed to balance speed, accuracy, and resource efficiency while providing rich, interconnected memories for intelligent retrieval and reasoning.

Structured Ingestion

Not all data needs the full 11-stage pipeline. Structured data — decisions, tool calls, plan tasks, code entities — already has its schema defined. Running it through LLM extraction is wasteful and lossy.

ingest_structured() bypasses the NLP pipeline entirely. A registered handler validates the data, converts it to a MemoryItem, declares edges and indexes, and controls whether an embedding is generated.

Three Strategies

Usage

# Structured decision (INDEXED — queried by status, domain)
memory.ingest_structured(
    {"content": "Use JWT for auth", "domain": "security", "status": "active"},
    schema="decision"
)

# Tool call telemetry (APPEND — recency + tool name filter)
memory.ingest_structured(
    {"tool_name": "grep", "result": "found 3 matches", "success": True},
    schema="tool_call"
)

# Seed pack item (FULL — reference knowledge with embedding)
memory.ingest_structured(
    {"content": "Python is a programming language", "memory_type": "semantic"},
    schema="seed_item"
)

Available Schemas

Custom Handlers

Create a handler by implementing the StructuredHandler protocol — no base class required:

class MyHandler:
    strategy = IngestionStrategy.INDEXED
    schema_name = "my_type"
    embed = None  # None = defer to strategy default

    def validate(self, data: dict) -> bool: ...
    def to_memory_item(self, data: dict) -> MemoryItem: ...
    def get_edges(self, data: dict, item_id: str) -> list[tuple]: ...
    def get_indexes(self) -> list[str]: ...

Register it via get_structured_registry().register("my_type", MyHandler()).