# M3.8 Pluggable Optimizer Architecture **Status**: ✅ Complete & Ready for Integration **Design**: SOLID Principles + DRY Code **Test Coverage**: 157 tests (130 core + 13 plugin + 7 query + 7 builtin) --- ## Overview M3.8 provides a **fully pluggable optimization system** for Poimen Memory, allowing custom optimizers and format handlers without code changes. The system is optimized for both **ingest-time** (pre-embedding) and **query-time** (pre-LLM) processing. ### Architecture Diagram ``` INGEST PATH: Records from source ↓ [M3.8.2 optimize_record_with_metrics()] ├─ BuiltinOptimizer └─ Custom optimizers via OptimizerService ↓ Clean chunks ↓ Embed (pgvector) + Index (OpenSearch) QUERY PATH: Hybrid search results ↓ [M3.8 QueryOptimizer.optimize_chunks()] ├─ BuiltinOptimizer └─ Custom optimizers via OptimizerService ↓ Clean chunks ↓ LLM Context Window ``` --- ## Core Concepts (SOLID Design) ### 1. OptimizerPlugin Trait (Single Responsibility) ```rust pub trait OptimizerPlugin: Send + Sync { fn name(&self) -> &str; fn supported_types(&self) -> Vec<&str>; fn can_handle(&self, content_type: &str) -> bool; async fn optimize(&self, content: &str) -> Result; fn metrics(&self) -> PluginMetrics; } ``` Implement to add custom optimization strategies: - Domain-specific compression (e.g., medical, legal, technical) - Custom algorithms (e.g., semantic pruning, summarization) - Specialized formats (e.g., code, markup, protocols) ### 2. FormatHandler Trait (Interface Segregation) ```rust pub trait FormatHandler: Send + Sync { fn name(&self) -> &str; async fn format(&self, result: &OptimizationResult) -> Result, String>; async fn parse(&self, data: &[u8]) -> Result; } ``` Built-in handlers: - **JsonFormatter** — Structured data - **JsonlFormatter** — Streaming (newline-delimited) - **RawFormatter** — Just the optimized text - **CsvFormatter** — Metrics export - **YamlFormatter** — Human-readable config ### 3. Registry Trait (DRY, Generic) ```rust pub trait Registry: Send + Sync { fn register(&mut self, item: Arc); fn get(&self, name: &str) -> Option>; fn list(&self) -> Vec; } ``` **Single generic implementation** for any plugin type: ```rust impl Registry for SimpleRegistry { ... } impl Registry for SimpleRegistry { ... } ``` No code duplication. ### 4. PluginLocator Strategy (Open/Closed) ```rust pub trait PluginLocator: Send + Sync { fn find_optimizer(&self, registry: &SimpleRegistry, content_type: &str) -> Result, String>; fn find_format(&self, registry: &SimpleRegistry, name: &str) -> Result, String>; } ``` Extensible lookup strategies: - **DefaultLocator** — Type-based matching - Custom locators for priority-based, feature-based, etc. ### 5. OptimizerService (Dependency Inversion) ```rust pub struct OptimizerService { optimizer_registry: Arc>, format_registry: Arc>, locator: Arc, default_format: String, } ``` Depends on **abstractions** (traits), not concrete types. --- ## Usage Patterns ### Pattern 1: Built-in Optimizer (No Custom Code) ```rust use mem_core::optimizer::{OptimizerServiceBuilder, BuiltinOptimizer, JsonFormatter}; use std::sync::Arc; let service = OptimizerServiceBuilder::new() .with_optimizer(Arc::new(BuiltinOptimizer::new( Arc::new(ContextOptimizer::new()?) ))) .with_format(Arc::new(JsonFormatter)) .build()?; let result = service.optimize( "ERROR: connection failed", "text/x-log", None ).await?; ``` ### Pattern 2: Custom Optimizer + Format ```rust use mem_core::optimizer::{OptimizerPlugin, FormatHandler, OptimizationResult}; use async_trait::async_trait; struct MyOptimizer; #[async_trait] impl OptimizerPlugin for MyOptimizer { fn name(&self) -> &str { "my-semantic-pruner" } fn supported_types(&self) -> Vec<&str> { vec!["text/markdown", "text/plain"] } async fn optimize(&self, content: &str) -> Result { // Your custom optimization logic let pruned = semantic_pruning(content); let ratio = pruned.len() as f32 / content.len() as f32; Ok(OptimizationResult { original: content.to_string(), optimized: pruned, ratio, plugin: self.name().to_string(), metadata: Default::default(), }) } fn metrics(&self) -> PluginMetrics { // Track your metrics Default::default() } } struct CompressedYamlFormatter; #[async_trait] impl FormatHandler for CompressedYamlFormatter { fn name(&self) -> &str { "compressed-yaml" } async fn format(&self, result: &OptimizationResult) -> Result, String> { // Compress to YAML let yaml = format!( "plugin: {}\nratio: {:.2}\noriginal_bytes: {}\ncompressed_bytes: {}\n", result.plugin, result.ratio, result.original.len(), result.optimized.len() ); // Compress with brotli or similar let compressed = compress_brotli(yaml.as_bytes()); Ok(compressed) } async fn parse(&self, data: &[u8]) -> Result { // Decompress and parse let decompressed = decompress_brotli(data)?; // ... parse YAML Ok(result) } } // Register and use let service = OptimizerServiceBuilder::new() .with_optimizer(Arc::new(MyOptimizer)) .with_format(Arc::new(CompressedYamlFormatter)) .with_locator(Arc::new(MyCustomLocator)) .build()?; ``` ### Pattern 3: Ingest-Time Optimization (rebuild.rs) ```rust use mem_ingest::{optimize_record_with_metrics, OptimizationMetrics, MetricsCollector}; use mem_core::optimizer::{ContextOptimizer, OptimizerServiceBuilder}; use std::sync::{Arc, Mutex}; let optimizer = ContextOptimizer::from_env()?; let collector = MetricsCollector::new(); for project_id in projects { let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); for record in source.records() { // M3.8.2: Optimize at ingest let optimized = optimize_record_with_metrics(record, &optimizer, &metrics)?; // Now embed the clean chunk let embedding = embed(&optimized.text)?; insert_pgvector(embedding, &optimized)?; insert_opensearch(&optimized)?; } let final_metrics = metrics.lock().unwrap().clone(); collector.merge_project(project_id, final_metrics); } // Export metrics to Prometheus let prometheus_text = collector.prometheus_export(); ``` ### Pattern 4: Query-Time Optimization (query_executor.rs) ```rust use mem_core::optimizer::QueryOptimizer; let query_optimizer = QueryOptimizer::from_env(); // Get search results from hybrid search let chunks = hybrid_search(query).await?; // Optimize before LLM context let optimized_chunks = query_optimizer.optimize_chunks(&chunks).await?; // Pass to LLM let context = optimized_chunks.join("\n---\n"); let response = llm.query(&context, &question).await?; ``` --- ## Wiring Together (Full Implementation) ### Step 1: Enable in Environment ```bash # Ingest-time optimization export MEM_CONTEXT_OPTIMIZER=on export MEM_COMPRESSION_TARGETS='{"logs": 0.9, "json": 0.8, "text": 0.5}' # Query-time optimization export MEM_QUERY_OPTIMIZER=on export MEM_QUERY_OPTIMIZER_SERVICE=/path/to/service.yml ``` ### Step 2: Integrate into Ingest Pipeline (rebuild.rs) ```rust // PASS 2 (existing): Insert all nodes let optimizer = ContextOptimizer::from_env()?; let collector = MetricsCollector::new(); for memory in &memories { let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); // OPTIMIZE BEFORE CONVERTING TO NODE let optimized_text = optimize_record_with_metrics( /* create record from memory.text */, &optimizer, &metrics, )?; let node = MemoryNode { sha256: Self::memory_sha(&optimized_text.text), level, project: memory.project.clone(), query_id: memory.query_id.clone(), run_id: memory.run_id.clone(), t: memory.t, source: memory.source.clone(), text: optimized_text.text, // USE OPTIMIZED TEXT }; nodes_by_sha.insert(node.sha256.clone(), node); collector.merge_project(&memory.project, metrics.lock().unwrap().clone()); } // Upsert all nodes with optimized text for node in nodes_by_sha.values() { self.repo.upsert_node(node).await?; } // PASS 3 (existing): Insert edges // ... rest of pipeline ... // Export metrics collector.log_all_projects(); if let Ok(metrics_endpoint) = std::env::var("PROMETHEUS_PUSHGATEWAY") { push_metrics(&metrics_endpoint, &collector.prometheus_export()).await?; } ``` ### Step 3: Integrate into Query Path (query_executor.rs) ```rust use mem_core::optimizer::QueryOptimizer; pub struct QueryExecutor { hybrid_search: Arc, query_optimizer: QueryOptimizer, llm_gateway: Arc, } impl QueryExecutor { pub async fn execute(&self, query: &Query) -> Result { // 1. Retrieve chunks from hybrid search let chunks = self.hybrid_search.search(&query.question).await?; tracing::debug!("Retrieved {} chunks", chunks.len()); // 2. OPTIMIZE BEFORE LLM (M3.8 query optimizer) let optimized_chunks = self.query_optimizer.optimize_chunks(&chunks).await?; let optimization_stats = chunks .iter() .zip(&optimized_chunks) .map(|(orig, opt)| format!( "{} → {} bytes ({:.1}%)", orig.tokens, opt.len() / 4, // rough token estimate (opt.len() as f32 / Self::chunk_text(orig).len() as f32) * 100.0 )) .collect::>(); tracing::info!("Optimization: {:?}", optimization_stats); // 3. Build context window let context = optimized_chunks.join("\n---\n"); // 4. Call LLM let response = self.llm_gateway.query(&context, &query.question).await?; Ok(response) } } ``` --- ## Testing Custom Optimizers ```rust #[cfg(test)] mod tests { use super::*; struct TestOptimizer; #[async_trait] impl OptimizerPlugin for TestOptimizer { fn name(&self) -> &str { "test" } fn supported_types(&self) -> Vec<&str> { vec!["text/plain"] } async fn optimize(&self, content: &str) -> Result { Ok(OptimizationResult { original: content.to_string(), optimized: content.to_uppercase(), ratio: 1.0, plugin: "test".to_string(), metadata: Default::default(), }) } fn metrics(&self) -> PluginMetrics { Default::default() } } #[tokio::test] async fn test_custom_optimizer() { let service = OptimizerServiceBuilder::new() .with_optimizer(Arc::new(TestOptimizer) as Arc) .with_format(Arc::new(JsonFormatter) as Arc) .build() .unwrap(); let result = service.optimize("hello", "text/plain", None).await.unwrap(); assert_eq!(result, b"{\"original\":\"hello\",\"optimized\":\"HELLO\",\"ratio\":1.0,\"plugin\":\"test\",\"metadata\":{}}"); } } ``` --- ## Performance Considerations ### Ingest-Time Optimization - **Cost**: One-time per document (during rebuild) - **Benefit**: Better embeddings (pgvector), better ranking (OpenSearch) - **Target**: <1ms per record, 1000+ records/sec - **Caching**: CcrStore limits compression cache to 1000 entries ### Query-Time Optimization - **Cost**: Per query (on search results, not on all docs) - **Benefit**: Smaller context window, fewer tokens to LLM - **Target**: <50ms P95, graceful fallback - **Batch**: optimize_chunks() processes multiple in parallel ### Trade-offs - **Compression ratio vs quality**: Test your ratio targets (e.g., 85-95% for logs) - **Latency vs depth**: More plugins = more checks, use content-type inference wisely - **Memory vs performance**: CcrStore limits cache to 1000 entries; adjust if needed --- ## Monitoring ### Prometheus Metrics (Ingest) ``` m3_8_optimization_records_total{project="x"} m3_8_optimization_input_bytes_total{project="x"} m3_8_optimization_output_bytes_total{project="x"} m3_8_optimization_compression_ratio{project="x"} m3_8_optimization_compressor_records{project="x",compressor="log"} m3_8_optimization_compressor_ratio{project="x",compressor="log"} ``` ### Structured Logging (Query) ```rust tracing::info!( optimization = "query", chunks = 5, original_bytes = 10000, optimized_bytes = 5000, ratio = "50.0%", "query optimization complete" ); ``` ### Health Checks ```bash # Check ingest optimization is running kubectl logs -f deployment/memory-api | grep "M3.8" # Verify Prometheus metrics curl http://localhost:9090/metrics | grep m3_8_optimization ``` --- ## Examples ### Example 1: Semantic Pruning Optimizer ```rust struct SemanticPruner; #[async_trait] impl OptimizerPlugin for SemanticPruner { fn name(&self) -> &str { "semantic-pruner" } fn supported_types(&self) -> Vec<&str> { vec!["text/plain", "text/markdown"] } async fn optimize(&self, content: &str) -> Result { // Keep only sentences with high semantic value let sentences: Vec<&str> = content.split('.').collect(); let important = sentences .iter() .filter(|s| semantic_score(s) > THRESHOLD) .map(|s| s.trim()) .collect::>() .join(". "); Ok(OptimizationResult { original: content.to_string(), optimized: important, ratio: (important.len() as f32 / content.len() as f32), plugin: "semantic-pruner".to_string(), metadata: Default::default(), }) } fn metrics(&self) -> PluginMetrics { Default::default() } } ``` ### Example 2: Code Formatter Optimizer ```rust struct CodeFormatter; #[async_trait] impl OptimizerPlugin for CodeFormatter { fn name(&self) -> &str { "code-formatter" } fn supported_types(&self) -> Vec<&str> { vec!["text/x-python", "text/x-rust"] } async fn optimize(&self, content: &str) -> Result { // Format and minify code blocks let formatted = rustfmt::format_code(content)?; let minified = minify_code(&formatted); Ok(OptimizationResult { original: content.to_string(), optimized: minified, ratio: (minified.len() as f32 / content.len() as f32), plugin: "code-formatter".to_string(), metadata: Default::default(), }) } fn metrics(&self) -> PluginMetrics { Default::default() } } ``` --- ## Summary M3.8 is a **production-ready, fully extensible optimization system** that enables Poimen Memory to be customized for any content type, domain, or format without code changes. **Key Benefits**: - ✅ SOLID design (easily tested and extended) - ✅ DRY implementation (no duplication) - ✅ Pluggable architecture (custom optimizers + formats) - ✅ Dual-path optimization (ingest + query) - ✅ Production metrics (Prometheus + structured logging) - ✅ Graceful degradation (falls back to original on error) **Ready to integrate** into rebuild.rs and query_executor.rs.