Background Processing

Source files

Key GitHub sources:

• automem/enrichment/runtime_queue_bindings.py — Enrichment queue setup
• automem/enrichment/runtime_worker.py — Enrichment worker thread
• automem/embedding/runtime_bindings.py — Embedding queue setup
• automem/embedding/runtime_pipeline.py — Embedding worker and batch processing
• automem/consolidation/runtime_bindings.py — Consolidation scheduler
• automem/sync/runtime_bindings.py — Sync worker
• automem/runtime_wiring.py — Startup sequence and worker initialization
• .env.example — Background processing configuration variables

AutoMem implements four background processing systems that operate independently of the main Flask API request/response cycle. These systems handle computationally expensive operations without blocking client requests.

For detailed implementation of each system, see:

• Enrichment Pipeline — Entity extraction and relationship building
• Embedding Generation — Batched vector creation

---

Four Independent Systems

AutoMem implements four background processing systems, each with distinct triggers, execution models, and responsibilities:

System	Trigger	Execution Model	Primary Purpose
Enrichment Pipeline	Event-driven	Queue + worker thread	Enhance memories with entities, tags, relationships
Embedding Worker	Event-driven	Queue + batch accumulator	Generate and store vector embeddings
Consolidation Engine	Time-based	Scheduled intervals	Apply decay, discover associations, cluster, forget
Sync Worker	Time-based	Interval polling	Detect and repair FalkorDB/Qdrant drift

---

Worker Coordination Diagram

graph TB
    subgraph write_ops["Write Operations"]
        PostMem["POST /memory<br/>api/memory.py"]
    end

    subgraph queues["Queue Management"]
        EnrichQueue["enrichment_queue<br/>ServiceState"]
        EmbedQueue["embedding_queue<br/>ServiceState"]

        EnrichPending["enrichment_pending<br/>Set[str]"]
        EnrichInflight["enrichment_inflight<br/>Set[str]"]
        EmbedPending["embedding_pending<br/>Set[str]"]
        EmbedInflight["embedding_inflight<br/>Set[str]"]
    end

    subgraph workers["Worker Threads"]
        direction TB

        EnrichWorker["EnrichmentWorker<br/>enrichment/runtime_worker.py"]
        EmbedWorker["EmbeddingWorker<br/>embedding/runtime_pipeline.py"]
        ConsolWorker["ConsolidationScheduler<br/>consolidation.py"]
        SyncWorker["SyncWorker<br/>sync/runtime_bindings.py"]
    end

    subgraph config["Configuration"]
        EnrichConfig["ENRICHMENT_IDLE_SLEEP_SECONDS=2<br/>ENRICHMENT_MAX_ATTEMPTS=3<br/>ENRICHMENT_SIMILARITY_LIMIT=5"]
        EmbedConfig["EMBEDDING_BATCH_SIZE=20<br/>EMBEDDING_BATCH_TIMEOUT_SECONDS=2.0"]
        ConsolConfig["CONSOLIDATION_TICK_SECONDS<br/>DECAY/CREATIVE/CLUSTER/FORGET<br/>_INTERVAL_SECONDS"]
        SyncConfig["SYNC_CHECK_INTERVAL_SECONDS<br/>SYNC_AUTO_REPAIR=true"]
    end

    PostMem --> EnrichQueue
    PostMem --> EmbedQueue

    EnrichQueue --> EnrichPending
    EmbedQueue --> EmbedPending

    EnrichWorker --> EnrichInflight
    EnrichWorker --> EnrichPending
    EmbedWorker --> EmbedInflight
    EmbedWorker --> EmbedPending

    EnrichConfig -.-> EnrichWorker
    EmbedConfig -.-> EmbedWorker
    ConsolConfig -.-> ConsolWorker
    SyncConfig -.-> SyncWorker

---

Queue Management

graph TB
    subgraph "HTTP Request Thread"
        PostMemory["POST /memory<br/>Route handler"]
    end

    subgraph "ServiceState Queues"
        EnrichQ["enrichment_queue: Queue<br/>Thread-safe FIFO"]
        EmbedQ["embedding_queue: Queue<br/>Thread-safe FIFO"]

        EnrichPending["enrichment_pending: Set<br/>Dedup tracking"]
        EnrichInflight["enrichment_inflight: Set<br/>Processing tracking"]

        EmbedPending["embedding_pending: Set<br/>Dedup tracking"]
        EmbedInflight["embedding_inflight: Set<br/>Processing tracking"]
    end

    subgraph "Background Worker Threads"
        EnrichWorker["enrichment_worker()<br/>Entity extraction thread"]
        EmbedWorker["embedding_worker()<br/>Batch embedding thread"]
    end

    PostMemory -->|"queue.put(memory_id)"| EnrichQ
    PostMemory -->|"queue.put(memory_id)"| EmbedQ

    EnrichQ --> EnrichPending
    EmbedQ --> EmbedPending

    EnrichPending -.->|"if not in pending/inflight"| EnrichWorker
    EmbedPending -.->|"if not in pending/inflight"| EmbedWorker

    EnrichWorker -->|"Processing"| EnrichInflight
    EmbedWorker -->|"Processing"| EmbedInflight

    EnrichInflight -.->|"Complete"| EnrichPending
    EmbedInflight -.->|"Complete"| EmbedPending

Key Insights:

• POST /memory writes immediately to FalkorDB, then enqueues background jobs
• Enrichment and embedding workers run continuously in separate threads
• Consolidation scheduler runs periodic checks via a custom thread-based scheduler
• All systems can operate independently; failures are isolated

---

Threading Model

All background workers run in daemon threads started during Flask application initialization:

Component	Started At	Daemon	Lifecycle
enrichment_worker	automem/enrichment/runtime_queue_bindings.py	Yes	Runs until app shutdown
embedding_worker	automem/embedding/runtime_bindings.py	Yes	Runs until app shutdown
Consolidation scheduler	automem/consolidation/runtime_bindings.py	Yes	Custom thread-based scheduler

Thread Safety:

• enrichment_queue and embedding_queue use Python’s thread-safe Queue class
• Each worker polls its queue in an infinite loop with timeout-based blocking
• FalkorDB and Qdrant clients are thread-safe for read/write operations

---

Lifecycle and Startup

Application Startup Sequence

Startup is orchestrated by automem/runtime_wiring.py:

1. init_falkordb() — Establish FalkorDB connection
2. init_qdrant() — Establish optional Qdrant connection
3. init_openai() — Initialize OpenAI client for classification
4. init_embedding_provider() — Select and initialize embedding provider
5. init_enrichment_pipeline() — Start enrichment worker thread
6. init_embedding_pipeline() — Start embedding worker thread
7. init_consolidation_scheduler() — Start consolidation scheduler
8. init_sync_worker() — Start sync worker thread

Graceful Shutdown

All background threads are daemon threads, meaning they terminate automatically when the main Flask process exits. No explicit cleanup is required.

Implications:

• In-flight enrichment/embedding jobs may be lost on shutdown
• Consolidation tasks may be interrupted mid-run
• Queue contents are not persisted between restarts
• Restarting the service will reprocess queued items from scratch (new memories will be re-enqueued on next access or re-enrichment trigger)

---

Configuration Reference

Enrichment Configuration

Variable	Default	Purpose
ENRICHMENT_MAX_ATTEMPTS	3	Max attempts per forced job (non-forced jobs are not retried)
ENRICHMENT_SIMILARITY_LIMIT	5	Max similar neighbors to link
ENRICHMENT_SIMILARITY_THRESHOLD	0.8	Min cosine similarity for SIMILAR_TO edge
ENRICHMENT_IDLE_SLEEP_SECONDS	2.0	Queue poll timeout
ENRICHMENT_FAILURE_BACKOFF_SECONDS	5.0	Flat sleep after a failure, before the re-enqueue
ENRICHMENT_ENABLE_SUMMARIES	true	Generate content summaries
ENRICHMENT_SPACY_MODEL	en_core_web_sm	spaCy model for NER

Embedding Configuration

Variable	Default	Purpose
EMBEDDING_BATCH_SIZE	20	Items per batch
EMBEDDING_BATCH_TIMEOUT_SECONDS	2.0	Max wait before processing partial batch

Consolidation Configuration

Variable	Default	Purpose
CONSOLIDATION_TICK_SECONDS	60	Scheduler check interval
CONSOLIDATION_DECAY_INTERVAL_SECONDS	86400	Decay task frequency (1 day)
CONSOLIDATION_CREATIVE_INTERVAL_SECONDS	604800	Creative task frequency (1 week)
CONSOLIDATION_CLUSTER_INTERVAL_SECONDS	2592000	Cluster task frequency (1 month)
CONSOLIDATION_FORGET_INTERVAL_SECONDS	0	Forget task frequency (0 = disabled)
CONSOLIDATION_DECAY_IMPORTANCE_THRESHOLD	0.3	Min importance for decay (optional filter)
CONSOLIDATION_HISTORY_LIMIT	20	Max consolidation run history

---

Monitoring and Health

Health Endpoint Statistics

The /health endpoint reports the enrichment pipeline’s queue state alongside store and sync status. It has no embedding or consolidation block:

Metric	Source	Interpretation
enrichment.status	ServiceState.enrichment_thread	running if the worker thread is alive, else stopped
enrichment.queue_depth	ServiceState.enrichment_queue	Items waiting in queue
enrichment.pending	ServiceState.enrichment_pending	Enqueued but not yet claimed by the worker; a job moves to enrichment.inflight when the worker picks it up
enrichment.inflight	ServiceState.enrichment_inflight	Currently processing
enrichment.processed	EnrichmentStats.successes	Total completed
enrichment.failed	EnrichmentStats.failures	Total failed
memory_count / vector_count	FalkorDB / Qdrant counts	Records in each store, excluding RECALL_EXCLUDED_TYPES; null when a store is unavailable
sync_status	Derived from the two counts	synced, drift_detected, orphaned_vectors, or unknown
vector_dimensions	VECTOR_SIZE, ServiceState.effective_vector_size, Qdrant collection	configured, effective, collection, mismatch

Embedding-queue depth is not published by /health. Consolidation scheduling and run history come from GET /consolidate/status, which returns next_runs, history, thread_alive, and tick_seconds.

Operational Endpoints

Monitoring endpoints and manual triggers. The global API-token guard covers all four whenever AUTOMEM_API_TOKEN is set (/health is exempt from it). Only /enrichment/reprocess also requires the admin token (ADMIN_API_TOKEN, sent in the X-Admin-Token or X-Admin-Api-Key header, or as ?admin_token); see Admin Operations.

Endpoint	Method	Auth	Purpose
/enrichment/status	GET	API token	Enrichment worker status and processing stats (no pending-ID samples)
/enrichment/reprocess	POST	API token + admin token	Re-enqueue specific memory IDs
/consolidate	POST	API token	Manually trigger consolidation tasks
/consolidate/status	GET	API token	Consolidation history and next run times

---

Error Handling and Retry Logic

Enrichment Retries

Only forced enrichment jobs (from POST /enrichment/reprocess) are retried, with a flat backoff:

• Max attempts: ENRICHMENT_MAX_ATTEMPTS (default: 3)
• Backoff: ENRICHMENT_FAILURE_BACKOFF_SECONDS (default: 5.0) — the worker sleeps this flat duration after each failure that is re-enqueued, then re-enqueues the job
• After the last attempt the worker logs Giving up on enrichment and does not re-queue the job

A non-forced job (the kind POST /memory queues) is not retried. The worker still sleeps the backoff after a failure that has attempts left, but the re-enqueue goes through the in-flight duplicate check and is dropped because the failed memory ID is still marked in flight. The worker does not retry it. POST /enrichment/reprocess queues the memory again.

Embedding Retries

Embedding failures are currently not retried at the batch level. If a batch fails:

1. Error is logged
2. Memories remain without embeddings in Qdrant
3. Graph metadata embedding_status remains in queue state
4. Re-embedding can be triggered via /admin/reembed

Consolidation Failures

Consolidation tasks catch exceptions and continue:

• Failed tasks are logged with error details
• Next scheduled run proceeds normally
• Individual task failures don’t affect other tasks
• History includes error details for debugging

---

Performance Characteristics

Resource Usage

System	CPU Usage	Memory Usage	I/O Pattern
Enrichment	Moderate (spaCy NER)	Low per job	Burst writes to graph
Embedding	Low (network-bound)	Low (batch accumulator)	Batched API calls
Consolidation	High (graph traversal)	Moderate (in-memory cache)	Large read + write operations
Sync Worker	Low (periodic scans)	Low	Periodic graph + vector queries

Optimization Features

Embedding Batching (40-50% cost reduction):

• Accumulates up to 20 memories before calling provider API
• Single API call generates embeddings for entire batch
• Timeout ensures responsiveness (max 2s delay)

Relationship Count Caching (80% consolidation speedup):

• LRU cache with 10,000 entry capacity
• Hourly cache invalidation via timestamp key
• Dramatically reduces graph queries during decay cycles (consolidation.py:201-216)

---

Integration with API Endpoints

POST /memory Integration

The POST /memory endpoint response includes indicators of queued background work:

{
  "status": "success",
  "memory_id": "uuid",
  "enrichment": "queued",
  "embedding_status": "queued"
}

GET /recall Integration

Recall queries benefit from completed background processing:

• Vector search uses embeddings generated by the embedding worker
• Relationship traversal uses edges created by the enrichment worker
• Relevance scores use relevance_score updated by the consolidation engine

Tip

Memories searched immediately after storage may not yet have embeddings or semantic relationships. The best search quality is achieved after the background workers have completed their processing (typically within 1-2 seconds for enrichment, 2-5 seconds for embedding depending on batch accumulation).

---

Summary

AutoMem’s background processing architecture achieves:

1. Non-blocking API responses — Write to graph first, enhance later
2. Cost efficiency — Batched embedding generation reduces API calls by 40-50%
3. Automatic maintenance — Scheduled consolidation keeps graph healthy over time
4. Fault tolerance — Independent systems with retry logic and error isolation
5. Observability — Rich metrics via /health and the operational endpoints

For implementation details of each subsystem:

• Enrichment Pipeline — Entity extraction, relationship building, pattern detection
• Embedding Generation — Provider selection, batching, dimension validation