feat: LLM entity + fact extraction pipeline (Zep paper alignment) (#48)

## Changes

### Entity Extraction
- Switch from WikiLinkFallbackExtractor to LlmEntityExtractor when LLM_ENDPOINT set
- `clean_llm_response()`: strips `<think>` tags, markdown fences, extracts JSON
- Handle array responses (Ollama returns `[...]` not `{entities: [...]}`)
- EntityType custom Deserialize: unknown variants → Unknown (no crash)
- Increase timeout 30s→90s, max_tokens 500→1500 for reasoning models
- Graceful reflection fallback: keep entities if verification fails

### Fact Extraction (NEW)
- LlmFactExtractor: LLM-based relationship extraction between entity pairs
- Validates source/target against known entity list (drops hallucinated edges)
- Same robust JSON cleaning for reasoning models + Ollama
- IngestWorker auto-selects LLM vs Simple based on LLM_ENDPOINT env

### K8s Deployment
- Add `command: ["/app/mem"]` (fix args replacing CMD)
- Add LLM_ENDPOINT, LLM_MODEL env vars for in-cluster LLM

## E2E Tested (local Ollama qwen2.5:3b)
- 12 entities extracted (person, tool, concept, organization)
- 5 edges with relationships and facts
- 781 tests pass

## Zep Paper Alignment (§2.2)
- Entity extraction + resolution (§2.2.1)
- Fact extraction between entity pairs (§2.2.2)
- Temporal edge invalidation ready (t_valid/t_invalid schema)
- Reflection verification (§2.2.1, graceful fallback)

---------

Co-authored-by: rock <[email protected]>
Reviewed-on: #48
Co-authored-by: poimen <[email protected]>

crates/mem-cli/src/cache_alignment.rs

@@ -224,9 +224,20 @@ impl KvCacheAligner {
 
     /// Pre-load hot chunks into cache
     pub fn preload_hot_chunks(&self, hot_chunks: Vec<(&str, &str)>) -> Result<()> {
+        let count = hot_chunks.len();
         for (chunk_id, text) in hot_chunks {
             self.cache.put(chunk_id, text);
         }
+        let metrics = self.cache.metrics();
+        tracing::info!(
+            target: "observability",
+            event = "cache_preload",
+            preloaded = count,
+            cache_hits = metrics.hits,
+            cache_misses = metrics.misses,
+            hit_ratio = format!("{:.2}", metrics.hit_ratio()),
+            "Cache preload complete"
+        );
         Ok(())
     }
 

crates/mem-cli/src/chunk_optimizer.rs

@@ -211,17 +211,32 @@ impl ChunkOptimizer {
 
     /// End-to-end optimization pipeline
     pub fn optimize(&self, chunks: Vec<OptimizableChunk>) -> (Vec<OptimizableChunk>, SelectionMetrics) {
+        let input_count = chunks.len();
+
         // Step 1: Filter by threshold
         let filtered = self.threshold_filter.filter(chunks.clone());
+        let after_filter = filtered.len();
 
         // Step 2: Deduplicate
         let (deduplicated, dedup_removed) = self.deduplicator.deduplicate(filtered);
+        let after_dedup = deduplicated.len();
 
         // Step 3: Select within budget
         let (selected, mut metrics) = self.budget_selector.select(deduplicated);
-
         metrics.dedup_removed = dedup_removed;
 
+        tracing::info!(
+            target: "observability",
+            event = "chunk_optimize",
+            input = input_count,
+            after_threshold_filter = after_filter,
+            after_dedup = after_dedup,
+            dedup_removed = dedup_removed,
+            selected = selected.len(),
+            budget_bytes = metrics.total_bytes,
+            "Chunk optimization complete"
+        );
+
         (selected, metrics)
     }
 }

crates/mem-cli/src/compaction.rs

@@ -346,7 +346,19 @@ pub async fn compact_memory(
     }
 
     total_stats.duration_ms = start.elapsed().as_millis() as u64;
-    info!("Compaction complete in {}ms: {:?}", total_stats.duration_ms, total_stats);
+    info!(
+        target: "observability",
+        event = "compaction_complete",
+        mode = ?mode,
+        duration_ms = total_stats.duration_ms,
+        duplicate_edges_deleted = total_stats.duplicate_edges_deleted,
+        stale_facts_deleted = total_stats.stale_facts_deleted,
+        semantic_merged = total_stats.semantic_merged,
+        llm_calls = total_stats.llm_calls,
+        bytes_freed = total_stats.bytes_freed,
+        human_reviews_queued = total_stats.human_reviews_queued,
+        "Compaction complete"
+    );
 
     Ok(total_stats)
 }

crates/mem-cli/src/full_pipeline.rs

@@ -344,6 +344,22 @@ impl FullPipeline {
 
         metrics.total_latency_ms = start.elapsed().as_millis() as u64;
 
+        tracing::info!(
+            target: "observability",
+            event = "full_pipeline_complete",
+            query = query,
+            candidates = metrics.wiki_scope_docs,
+            prefiltered = metrics.prefilter_candidates,
+            optimized = metrics.post_optimization_count,
+            dedup_removed = metrics.dedup_removed,
+            boosts_applied = metrics.metadata_boosts_applied,
+            cache_hit_ratio = format!("{:.2}", metrics.cache_hit_ratio),
+            budget_bytes = metrics.budget_used_bytes,
+            total_ms = metrics.total_latency_ms,
+            "Full query pipeline complete"
+        );
+
+
         Ok(PipelineResult {
             query: query.to_string(),
             query_intent,
@@ -467,6 +483,22 @@ impl FullPipeline {
 
         metrics.total_latency_ms = start.elapsed().as_millis() as u64;
 
+        tracing::info!(
+            target: "observability",
+            event = "full_pipeline_complete",
+            query = query,
+            candidates = metrics.wiki_scope_docs,
+            prefiltered = metrics.prefilter_candidates,
+            optimized = metrics.post_optimization_count,
+            dedup_removed = metrics.dedup_removed,
+            boosts_applied = metrics.metadata_boosts_applied,
+            cache_hit_ratio = format!("{:.2}", metrics.cache_hit_ratio),
+            budget_bytes = metrics.budget_used_bytes,
+            total_ms = metrics.total_latency_ms,
+            "Full query pipeline complete"
+        );
+
+
         Ok(PipelineResult {
             query: query.to_string(),
             query_intent,

crates/mem-cli/src/ingest_worker.rs

@@ -2,14 +2,15 @@ use anyhow::Result;
 use mem_store::{MemoryL1, VectorStore, ChunkL0, EntityRepoOps, EdgeRepoOps};
 use mem_llm::EmbeddingsClient;
 use mem_ingest::ingest_pipeline::{IngestPipeline, Episode};
-use mem_ingest::entity_extractor::WikiLinkFallbackExtractor;
-use mem_ingest::fact_extractor::SimpleFactExtractor;
+use mem_ingest::entity_extractor::{WikiLinkFallbackExtractor, LlmEntityExtractor};
+use mem_ingest::fact_extractor::{SimpleFactExtractor, LlmFactExtractor};
 use mem_ingest::contradiction_detector::ContradictionHandler;
 use sqlx::PgPool;
 use uuid::Uuid;
 use std::sync::Arc;
 use pgvector::Vector;
 
+
 /// Ingest worker — processes queued records through entity/fact extraction pipeline
 pub struct IngestWorker {
     pool: PgPool,
@@ -26,11 +27,25 @@ impl IngestWorker {
     ) -> Self {
         let vector_store = Arc::new(VectorStore::new(pool.clone()));
         
-        // Initialize extraction pipeline
+        // Initialize extraction pipeline — use LLM if LLM_ENDPOINT is set, else fallback to wiki links
         let entity_extractor: Arc<dyn mem_ingest::entity_extractor::EntityExtractor> = 
-            Arc::new(WikiLinkFallbackExtractor);
+            if std::env::var("LLM_ENDPOINT").is_ok() {
+                let model = std::env::var("LLM_MODEL").unwrap_or_else(|_| "qwen2.5:3b-instruct".to_string());
+                tracing::info!("Using LLM entity extractor: model={}", model);
+                Arc::new(LlmEntityExtractor::new(&model))
+            } else {
+                tracing::info!("LLM_ENDPOINT not set, using WikiLink fallback extractor");
+                Arc::new(WikiLinkFallbackExtractor)
+            };
         let fact_extractor: Arc<dyn mem_ingest::fact_extractor::FactExtractor> = 
-            Arc::new(SimpleFactExtractor);
+            if std::env::var("LLM_ENDPOINT").is_ok() {
+                let model = std::env::var("LLM_MODEL").unwrap_or_else(|_| "qwen2.5:3b-instruct".to_string());
+                tracing::info!("Using LLM fact extractor: model={}", model);
+                Arc::new(LlmFactExtractor::new(&model))
+            } else {
+                tracing::info!("LLM_ENDPOINT not set, using simple pattern fact extractor");
+                Arc::new(SimpleFactExtractor)
+            };
         let contradiction_detector = Arc::new(ContradictionHandler::default());
         let pipeline = Arc::new(IngestPipeline::new(
             entity_extractor,
@@ -121,9 +136,15 @@ impl IngestWorker {
             .await?;
 
         tracing::info!(
-            "Ingest completed: {} (entities={}, edges={}, reviews={})",
-            ingest_id, total_entities, total_edges, total_reviews
+            target: "observability",
+            event = "ingest_complete",
+            ingest_id = ingest_id,
+            entities = total_entities,
+            edges = total_edges,
+            reviews = total_reviews,
+            "Ingest completed"
         );
+
         Ok(())
     }
 
@@ -173,7 +194,12 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) ->
     sqlx::query(
         "INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence)
          VALUES ($1, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8)
-         ON CONFLICT (id) DO NOTHING"
+         ON CONFLICT (project_id, name) DO UPDATE SET
+           entity_type = EXCLUDED.entity_type,
+           description = COALESCE(NULLIF(EXCLUDED.description, ''), memory_entity.description),
+           t_updated = NOW(),
+           confidence = GREATEST(memory_entity.confidence, EXCLUDED.confidence),
+           source_count = memory_entity.source_count + 1"
     )
     .bind(&entity.id)
     .bind(&entity.project_id)
@@ -193,7 +219,7 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) ->
 async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<()> {
     // Try temporal schema first (id, project_id, source_entity_id, etc)
     let result = sqlx::query(
-        "INSERT INTO memory_edge (id, project_id, source_entity_id, target_entity_id, relation_type, fact, t_valid, t_invalid, t_created, confidence)
+        "INSERT INTO memory_edge (id, project_id, source_id, target_id, relation_type, fact, t_valid, t_invalid, t_created, confidence)
          VALUES ($1, $2, $3, $4, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10)
          ON CONFLICT (id) DO NOTHING"
     )

crates/mem-cli/src/query_router.rs

@@ -238,6 +238,17 @@ impl QueryRouter {
 
         let latency_ms = start.elapsed().as_millis() as u64;
 
+        tracing::info!(
+            target: "observability",
+            event = "query_route",
+            route = "direct",
+            candidates = all_candidates.len(),
+            prefiltered = prefilter_size,
+            selected = selected_chunks.len(),
+            latency_ms = latency_ms,
+            "Query routing complete"
+        );
+
         Ok(RoutedResult {
             selected_chunks,
             route,

crates/mem-cli/src/result_compressor.rs

@@ -235,6 +235,18 @@ impl BudgetCompressor {
         let strategy = self.select_strategy(estimated);
         let compressed = self.compressor.compress_batch(results, strategy);
 
+        let compressed_size: usize = compressed.iter().map(|c| c.text.as_ref().map_or(0, |t| t.len())).sum();
+        tracing::info!(
+            target: "observability",
+            event = "result_compress",
+            input_count = compressed.len(),
+            estimated_bytes = estimated,
+            compressed_bytes = compressed_size,
+            budget_bytes = self.max_budget_bytes,
+            strategy = ?strategy,
+            "Result compression complete"
+        );
+
         (compressed, strategy)
     }
 }