Enrichment Pipeline

The Enrichment Pipeline is a background worker system that automatically enhances stored memories with extracted entities, relationships, summaries, and pattern associations. This page documents the queue-based architecture, processing stages, entity extraction techniques, and relationship creation mechanisms.

For information about embedding generation (which runs in parallel), see Embedding Generation. For details about consolidation cycles (decay, creative, cluster, forget), see the Consolidation Engine page. For the broader context of all background processing systems, see Background Processing.

System Architecture

The enrichment pipeline operates as an independent background thread that consumes jobs from a thread-safe queue. It processes memories asynchronously after storage, ensuring API write operations remain non-blocking.

Processing Flow

sequenceDiagram
    participant API as Flask API
    participant Queue as enrichment_queue
    participant Worker as Enrichment Worker
    participant Graph as FalkorDB
    participant Vector as Qdrant

    API->>Queue: enqueue(EnrichmentJob)
    Note over Queue: Add to enrichment_pending

    Worker->>Queue: dequeue (blocking)
    Note over Worker: Move to enrichment_inflight

    Worker->>Graph: MATCH memory by ID
    Worker->>Worker: extract_entities(content)
    Worker->>Worker: generate_summary(content)

    Worker->>Graph: MATCH recent memories<br/>CREATE PRECEDED_BY edges

    Worker->>Vector: search(embedding, limit=5, threshold=0.8)
    Vector-->>Worker: similar_memories
    Worker->>Graph: CREATE SIMILAR_TO edges<br/>SET strength = cosine_score

    Worker->>Graph: MATCH memories by type<br/>Detect pattern keywords
    Worker->>Graph: MERGE Pattern node<br/>CREATE EXEMPLIFIES edges

    Worker->>Graph: SET enriched=true<br/>SET enriched_at=timestamp<br/>SET metadata.enrichment

    Note over Worker: Remove from enrichment_inflight
    Worker->>Queue: record_success

Data Structures

EnrichmentJob

Jobs enqueued for processing contain the memory ID, retry attempt counter, and a forced flag for admin-triggered reprocessing.

Fields:

memory_id — UUID of the memory to enrich
attempt — Attempt counter (0-indexed), incremented when a forced job is re-enqueued after a failure
forced — When true (admin trigger), skips the already-enriched check and reprocesses

(automem/service_state.py)

Job Lifecycle

Job Creation

Enrichment jobs are enqueued in three scenarios:

Trigger	Endpoint	Behavior
Memory creation	`POST /memory`	Automatic enqueue after graph write
Batch creation	`POST /memory/batch`	Automatic enqueue for each stored memory
Admin reprocessing	`POST /enrichment/reprocess`	Forced reprocessing with `forced=true`

PATCH /memory/:id does not enqueue enrichment, so an edit does not refresh a memory’s entities, summary, or links. Run POST /enrichment/reprocess to re-enrich it.

Retry Logic

Only forced jobs are retried. After a failure that has attempts left, the worker sleeps ENRICHMENT_FAILURE_BACKOFF_SECONDS (a flat delay, 5 seconds by default) and calls enqueue_enrichment again with the attempt counter incremented. That call goes through the same duplicate check as a new job, and the failed memory ID is still marked in flight, so a non-forced job (the kind POST /memory queues) is not re-queued. Forced jobs from POST /enrichment/reprocess skip the check and are re-queued until ENRICHMENT_MAX_ATTEMPTS is used up:

Attempt (forced job)	Backoff before it	Behavior
1	0s	Immediate first attempt
2	`ENRICHMENT_FAILURE_BACKOFF_SECONDS`	Default 5 seconds
3	`ENRICHMENT_FAILURE_BACKOFF_SECONDS`	Default 5 seconds
Final	—	Worker logs `Giving up on enrichment`; each failure is already counted in `enrichment_stats`

A non-forced job gets one attempt. The worker still sleeps the backoff after a failure that has attempts left, but the re-enqueue is dropped, so the worker does not retry it. POST /enrichment/reprocess queues the memory again.

stateDiagram-v2
    [*] --> Queued
    Queued --> Processing: dequeue
    Processing --> Success: enrichment complete
    Processing --> Retry: error (forced job, attempts left)
    Processing --> Failed: error (non-forced job, or no attempts left)

    Retry --> Queued: re-enqueue with attempt++
    Success --> [*]
    Failed --> [*]: record_failure

Entity Extraction

Extraction Methods

The pipeline uses a two-tier approach: spaCy NLP when available, with regex fallbacks for specific patterns.

Tier 1: spaCy NLP

Runs en_core_web_sm model (configurable via ENRICHMENT_SPACY_MODEL)
Model is loaded on first use into the module-level _SPACY_NLP global, guarded by _SPACY_INIT_LOCK (a threading.Lock), and reused for later extractions
Named Entity Recognition (NER) extracts PERSON, ORG, PRODUCT, WORK_OF_ART, LAW, EVENT, GPE, LOC, NORP labels

Tier 2: Regex Fallbacks

Active when spaCy is unavailable or as supplements for patterns spaCy misses
Patterns like "met with X", "talked to X" for people
Backtick-delimited terms for project names
"using X", "deploy X" patterns for tools

Entity Types

Five entity categories are extracted and stored in metadata.entities:

Type	spaCy Labels	Regex Patterns	Examples
`people`	`PERSON`	`met with X`, `talked to X`	”Sarah”, “John Smith”
`organizations`	`ORG`	—	“Google”, “NASA”
`tools`	`PRODUCT`, `WORK_OF_ART`, `LAW`	`using X`, `deploy X`	”PostgreSQL”, “Docker”
`concepts`	`EVENT`, `GPE`, `LOC`, `NORP`	—	“Machine Learning”, “New York”
`projects`	—	backticks, `project called "X"`	”automem”, “Project Phoenix”

Entity Validation

The _is_valid_entity function filters noise using multiple heuristics.

Rejection criteria:

Length < 3 characters
Matches SEARCH_STOPWORDS, ENTITY_STOPWORDS, or ENTITY_BLOCKLIST
No alphabetic characters
Starts lowercase (unless allow_lower=true)
Starts with markdown/code symbols (-, *, #, `, {, etc.)
Ends with class name suffixes (Adapter, Handler, Manager, Service, etc.)
Boolean/null literals (true, false, null, undefined)
Environment variable pattern (UPPER_CASE with underscores)
Exceeds max_words limit (if specified)

Auto-Tagging

Extracted entities generate structured tags in the format entity:<type>:<slug>:

"John Smith" → entity:people:john-smith
"PostgreSQL" → entity:tools:postgresql
"Project Phoenix" → entity:projects:project-phoenix

Slugification uses _slugify to convert entities to URL-safe lowercase identifiers.

Summary Generation

The generate_summary function creates lightweight gist representations for quick scanning, controlled by the ENRICHMENT_ENABLE_SUMMARIES configuration flag.

Algorithm

flowchart LR
    Content["Memory Content"]
    Split["Split into sentences<br/>regex: (?<=[.!?])\\s+"]
    First["Take first sentence"]
    Truncate["Truncate to max_length<br/>Default: 240 chars"]
    Trim["Trim at last word boundary<br/>rsplit(' ', 1)"]
    Summary["Summary"]

    Content --> Split
    Split --> First
    First --> Truncate
    Truncate --> Trim
    Trim --> Summary

Example:

Input: "The team decided to migrate the database to PostgreSQL. This was due to performance concerns. Migration planned for Q2."

Output: "The team decided to migrate the database to PostgreSQL."

Relationship Creation

The enrichment pipeline automatically creates three types of relationships to build the knowledge graph.

Temporal Links (PRECEDED_BY)

Connects a memory to up to 5 earlier memories from the last 7 days.

Queries FalkorDB for memories whose timestamp is earlier than this memory’s and falls within the last 7 days (a fixed cutoff, not configurable), newest first, up to 5
Creates PRECEDED_BY edges from the new memory to those recent memories
Establishes chronological chains in the graph

Semantic Links (SIMILAR_TO)

Creates bidirectional edges between semantically similar memories using Qdrant vector search.

Semantic Linking Flow:

Query Qdrant for the memory’s embedding (if available)
Search for similar vectors above the similarity threshold
Create SIMILAR_TO edges in FalkorDB with strength = cosine_score
Links are bidirectional (both directions created)

Configuration:

ENRICHMENT_SIMILARITY_LIMIT — Maximum neighbors to link (default: 5)
ENRICHMENT_SIMILARITY_THRESHOLD — Minimum cosine similarity (default: 0.8)

Pattern Detection (EXEMPLIFIES)

Discovers recurring themes by analyzing memories of the same type and linking them to shared Pattern nodes.

Pattern Detection Algorithm:

Load recent memories of the same type
Extract key term frequencies (TF-IDF-style counting)
Identify terms appearing in multiple memories above threshold
MERGE a Pattern node identified by slugified theme name
Create EXEMPLIFIES edge from memory to pattern node

Pattern node properties:

pattern_id — Unique identifier (slugified theme)
occurrences — Counter incremented each time the pattern is detected
first_seen — Timestamp of first occurrence
last_reinforced — Timestamp of most recent occurrence

EXEMPLIFIES relationship properties:

pattern_type — Memory type that exemplifies the pattern
confidence — Detection confidence score
key_terms — Array of keywords that define the pattern

Configuration

Environment Variables

Variable	Default	Description
`ENRICHMENT_ENABLE_SUMMARIES`	`true`	Enable automatic summary generation
`ENRICHMENT_MAX_ATTEMPTS`	`3`	Maximum attempts for a forced job before giving up (non-forced jobs are not retried)
`ENRICHMENT_IDLE_SLEEP_SECONDS`	`2`	Worker sleep duration when queue is empty
`ENRICHMENT_FAILURE_BACKOFF_SECONDS`	`5`	Flat delay the worker sleeps after a failure, before it re-enqueues
`ENRICHMENT_CIRCUIT_COOLDOWN_SECONDS`	`300`	Cooldown (seconds) after a definitive LLM quota-exhaustion error before one recovery probe is allowed
`ENRICHMENT_SIMILARITY_LIMIT`	`5`	Maximum semantic neighbors to link
`ENRICHMENT_SIMILARITY_THRESHOLD`	`0.8`	Minimum cosine similarity for `SIMILAR_TO` edges
`ENRICHMENT_SPACY_MODEL`	`en_core_web_sm`	spaCy model for NER (requires `pip install spacy`)

Monitoring and Observability

Status Endpoint

The GET /enrichment/status endpoint exposes real-time worker metrics:

{
  "status": "running",
  "queue_size": 3,
  "pending": 3,
  "inflight": 1,
  "max_attempts": 3,
  "stats": {
    "processed_total": 1024,
    "successes": 1021,
    "failures": 3,
    "last_success_id": "a1b2c3d4-...",
    "last_success_at": "2025-01-15T10:30:00Z",
    "last_error": "FalkorDB write failed",
    "last_error_at": "2025-01-14T08:15:00Z"
  },
  "classification": {
    "llm_attempts": 512,
    "llm_successes": 510,
    "fallbacks": 2,
    "pattern_classifications": 502,
    "last_error": null,
    "last_error_at": null
  },
  "circuit": {
    "open": false,
    "circuit_open_skips": 0,
    "recoveries": 0
  }
}

Fields:

status — Worker thread state: "running" or "stopped"
queue_size — Items in the underlying queue (enrichment_queue.qsize())
pending — Count of IDs in enrichment_pending (items not yet dequeued; len(state.enrichment_pending))
inflight — Count of IDs in enrichment_inflight (items currently processing; len(state.enrichment_inflight))
max_attempts — Configured attempt limit for forced jobs (ENRICHMENT_MAX_ATTEMPTS)
stats — Lifetime counters from EnrichmentStats.to_dict(): processed_total, successes, failures, last_success_id, last_success_at, last_error, last_error_at
classification — LLM type-classification counters: llm_attempts, llm_successes, fallbacks (LLM called but fell back to default), pattern_classifications (resolved by regex patterns without LLM), last_error, last_error_at. Monitors how often LLM classification is used vs. pattern matching, and the fallback rate.
circuit — Quota-circuit state: open (true when the circuit is tripped and LLM calls are paused), circuit_open_skips (total LLM calls skipped while circuit was open), recoveries (total times the recovery probe succeeded and the circuit closed). Controlled by ENRICHMENT_CIRCUIT_COOLDOWN_SECONDS.

Tracking Sets

Thread-safe sets prevent duplicate processing:

enrichment_pending — Set of memory IDs awaiting enrichment (not yet dequeued)
enrichment_inflight — Set of memory IDs currently being processed

When a job is dequeued, the ID is moved from enrichment_pending to enrichment_inflight. It is removed from enrichment_inflight when the attempt ends, after any retry re-enqueue has been attempted, so a failed memory is still in flight when the worker re-enqueues it. Non-forced enqueues check both sets and are dropped when the ID is present; forced enqueues skip the check.

Error Handling

Failure Scenarios

Scenario	Behavior	Recovery
spaCy model unavailable	Log warning, use regex fallbacks	Graceful degradation
FalkorDB write failure	Forced jobs retry with flat backoff; non-forced jobs are not retried	Up to `ENRICHMENT_MAX_ATTEMPTS` attempts (forced jobs)
Qdrant unavailable	Skip semantic linking, log warning	Continue enrichment
Memory ID not found	`enrich_memory` returns `False` and logs at debug level; the worker records a skipped success, not a failure	No retry
Extraction exception	Catch and log; forced jobs retry, non-forced jobs are not retried	Up to `ENRICHMENT_MAX_ATTEMPTS` attempts (forced jobs)

Retry Mechanics

Forced jobs increment their attempt counter on each retry. A non-forced job is never re-queued, because the in-flight duplicate check drops its re-enqueue. When a forced job’s attempt + 1 reaches ENRICHMENT_MAX_ATTEMPTS, the worker logs Giving up on enrichment and does not re-queue it.

The backoff is a flat sleep of ENRICHMENT_FAILURE_BACKOFF_SECONDS after each failure that is re-enqueued. For a forced job with the defaults:

Attempt 1 → immediate
Attempt 2 → sleep 5 seconds
Attempt 3 → sleep 5 seconds
After 3 failures → give up (each failure is already counted in enrichment_stats)

Performance Characteristics

Throughput

Entity extraction: ~50-100ms per memory (spaCy), ~5-10ms (regex only)
Summary generation: <1ms (single sentence extraction)
Temporal linking: ~10-50ms (depends on time window size)
Semantic linking: ~20-100ms (Qdrant query for 5 neighbors)
Pattern detection: ~50-200ms (depends on memory type frequency)

Total enrichment time per memory: ~150-500ms

Concurrency

Single worker thread processes jobs sequentially
Non-blocking API writes (enrichment happens asynchronously)
Lock-protected tracking sets prevent race conditions
Queue-based architecture allows future multi-worker scaling

Admin Operations

Force Reprocessing

The POST /enrichment/reprocess endpoint (requires X-Admin-Token) allows forced re-enrichment of existing memories:

Parameters:

ids — Memory UUIDs to reprocess, as a JSON array or a comma-separated string in the JSON body. The comma-separated ?ids= query parameter is used only when the body has no ids value. IDs are trimmed and de-duplicated, and blank entries and non-string array items are ignored.

Responses (once the admin-token check passes):

202 Accepted — {"status": "queued", "count": <n>, "ids": [<sorted ids>]}, where count is the number of distinct IDs; each one is enqueued with forced=true
400 Bad Request — no usable IDs were supplied; the body is {"status": "error", "code": 400, "message": "No memory ids provided for reprocessing"}

Integration with Other Systems

Relationship to Embedding Worker

Enrichment and embedding generation are independent workers:

System	Trigger	Dependency
Embedding Worker	Memory created	None (immediate queue)
Enrichment Worker	Memory created	Uses Qdrant for similarity search (uses existing embeddings if available)

Both can proceed in parallel. Enrichment can create SIMILAR_TO edges even if embedding generation is still in progress, using whatever embeddings are already in Qdrant.

For details on embedding generation, see Embedding Generation.

Relationship to Consolidation

Enrichment occurs immediately after memory creation, while consolidation runs on scheduled intervals:

Enrichment: Per-memory processing (immediate, within seconds)
Consolidation: Cross-memory analysis (daily/weekly/monthly intervals)

Enrichment creates the foundation (entities, initial relationships) that consolidation builds upon (decay scores, creative associations, clustering).

For details on consolidation cycles, see the Consolidation Engine page.