feat: add ctxgraph-embed + ctxgraph-mcp with RRF fusion search (v0.3)

Author: rohansx

Cargo.lock

@@ -3,7 +3,7 @@ resolver = "2"
 members = ["crates/*"]
 
 [workspace.package]
-version = "0.2.0"
+version = "0.3.0"
 edition = "2024"
 license = "MIT"
 repository = "https://github.com/rohansx/ctxgraph"
@@ -22,3 +22,4 @@ toml = "0.8"
 gline-rs = "1.0"
 ort = "=2.0.0-rc.9"
 ndarray = "0.16"
+fastembed = "4"

Cargo.toml

@@ -90,8 +90,6 @@ Output: Entities  → Postgres (Component), SQLite (Component), billing (Service
 
 Entity types and relation labels are fully configurable via `ctxgraph.toml`. Define what matters to your domain — Person, Component, Decision, Policy, whatever fits.
 
-For higher accuracy, optional tiers add coreference resolution, fuzzy dedup, and LLM-powered contradiction detection. Each tier is additive — Tier 1 alone covers 85%+ of structured text.
-
 ### Bi-Temporal History
 
 Every relationship tracks two time dimensions:

README.md

@@ -11,7 +11,7 @@ name = "ctxgraph"
 path = "src/main.rs"
 
 [dependencies]
-ctxgraph = { version = "0.2.0", path = "../ctxgraph-core" }
+ctxgraph = { version = "0.3.0", path = "../ctxgraph-core" }
 clap = { workspace = true }
 serde_json = { workspace = true }
 chrono = { workspace = true }

crates/ctxgraph-cli/Cargo.toml

@@ -14,7 +14,7 @@ thiserror = { workspace = true }
 uuid = { workspace = true }
 rusqlite = { workspace = true }
 toml = { workspace = true }
-ctxgraph-extract = { path = "../ctxgraph-extract", version = "0.2.0", optional = true }
+ctxgraph-extract = { path = "../ctxgraph-extract", version = "0.3.0", optional = true }
 
 [features]
 default = ["extract"]

crates/ctxgraph-core/Cargo.toml

@@ -23,6 +23,9 @@ pub enum CtxGraphError {
     #[error("extraction error: {0}")]
     Extraction(String),
 
+    #[error("embed error: {0}")]
+    Embed(String),
+
     #[error("io error: {0}")]
     Io(#[from] std::io::Error),
 }

crates/ctxgraph-core/src/error.rs

@@ -256,6 +256,113 @@ impl Graph {
             .link_episode_entity(episode_id, entity_id, span_start, span_end)
     }
 
+    // ── Embeddings ──
+
+    /// Store an embedding for an episode. The embedding is serialized as
+    /// little-endian f32 bytes.
+    pub fn store_embedding(&self, episode_id: &str, embedding: &[f32]) -> Result<()> {
+        let bytes: Vec<u8> = embedding
+            .iter()
+            .flat_map(|f| f.to_le_bytes())
+            .collect();
+        self.storage.store_episode_embedding(episode_id, &bytes)
+    }
+
+    /// Store an embedding for an entity.
+    pub fn store_entity_embedding(&self, entity_id: &str, embedding: &[f32]) -> Result<()> {
+        let bytes: Vec<u8> = embedding
+            .iter()
+            .flat_map(|f| f.to_le_bytes())
+            .collect();
+        self.storage.store_entity_embedding(entity_id, &bytes)
+    }
+
+    /// Load all episode embeddings as (episode_id, Vec<f32>) pairs.
+    pub fn get_embeddings(&self) -> Result<Vec<(String, Vec<f32>)>> {
+        let raw = self.storage.get_all_episode_embeddings()?;
+        let result = raw
+            .into_iter()
+            .map(|(id, bytes)| {
+                let floats: Vec<f32> = bytes
+                    .chunks_exact(4)
+                    .map(|c| f32::from_le_bytes(c.try_into().unwrap()))
+                    .collect();
+                (id, floats)
+            })
+            .collect();
+        Ok(result)
+    }
+
+    /// Fused search using Reciprocal Rank Fusion (RRF) over FTS5 + semantic results.
+    ///
+    /// `query_embedding` should be the pre-computed embedding for `query`.
+    /// Returns episodes ranked by combined RRF score.
+    pub fn search_fused(
+        &self,
+        query: &str,
+        query_embedding: &[f32],
+        limit: usize,
+    ) -> Result<Vec<FusedEpisodeResult>> {
+        const K: f64 = 60.0;
+
+        // Accumulate RRF scores per episode id
+        let mut scores: std::collections::HashMap<String, f64> = std::collections::HashMap::new();
+        let mut episodes_map: std::collections::HashMap<String, Episode> =
+            std::collections::HashMap::new();
+
+        // --- FTS5 ranked list ---
+        // Fetch a generous pool for RRF (up to 10x limit or 200)
+        let fts_pool = (limit * 10).max(200);
+        let fts_results = self.storage.search_episodes(query, fts_pool);
+        if let Ok(fts) = fts_results {
+            for (rank, (episode, _)) in fts.into_iter().enumerate() {
+                let rrf = 1.0 / (K + rank as f64 + 1.0);
+                *scores.entry(episode.id.clone()).or_insert(0.0) += rrf;
+                episodes_map.insert(episode.id.clone(), episode);
+            }
+        }
+
+        // --- Semantic (cosine similarity) ranked list ---
+        let all_embeddings = self.get_embeddings()?;
+        if !all_embeddings.is_empty() && !query_embedding.is_empty() {
+            // Compute cosine similarities
+            let mut semantic: Vec<(String, f32)> = all_embeddings
+                .into_iter()
+                .map(|(id, vec)| {
+                    let sim = cosine_similarity(query_embedding, &vec);
+                    (id, sim)
+                })
+                .collect();
+            // Sort descending by similarity
+            semantic.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
+
+            for (rank, (ep_id, _sim)) in semantic.into_iter().enumerate() {
+                let rrf = 1.0 / (K + rank as f64 + 1.0);
+                *scores.entry(ep_id.clone()).or_insert(0.0) += rrf;
+                // Fetch episode if not already cached
+                if !episodes_map.contains_key(&ep_id) {
+                    if let Ok(Some(ep)) = self.storage.get_episode(&ep_id) {
+                        episodes_map.insert(ep_id, ep);
+                    }
+                }
+            }
+        }
+
+        // Sort by total RRF score descending
+        let mut fused: Vec<(String, f64)> = scores.into_iter().collect();
+        fused.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
+
+        let results = fused
+            .into_iter()
+            .take(limit)
+            .filter_map(|(id, score)| {
+                episodes_map.remove(&id).map(|episode| FusedEpisodeResult { episode, score })
+            })
+            .collect();
+
+        Ok(results)
+    }
+
     // ── Search ──
 
     /// Search episodes via FTS5 full-text search.
@@ -319,3 +426,19 @@ impl Graph {
         self.storage.stats()
     }
 }
+
+/// Compute cosine similarity between two f32 vectors.
+/// Returns 0.0 if either vector has zero magnitude.
+fn cosine_similarity(a: &[f32], b: &[f32]) -> f32 {
+    if a.len() != b.len() || a.is_empty() {
+        return 0.0;
+    }
+    let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
+    let mag_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
+    let mag_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
+    if mag_a == 0.0 || mag_b == 0.0 {
+        0.0
+    } else {
+        dot / (mag_a * mag_b)
+    }
+}

crates/ctxgraph-core/src/graph.rs

@@ -2,7 +2,8 @@ use rusqlite::Connection;
 
 use crate::error::Result;
 
-const MIGRATIONS: &[(&str, &str)] = &[(
+const MIGRATIONS: &[(&str, &str)] = &[
+(
     "001_initial",
     r#"
     -- Episodes: raw events
@@ -144,6 +145,13 @@ const MIGRATIONS: &[(&str, &str)] = &[(
         VALUES (new.rowid, new.fact, new.relation);
     END;
     "#,
+),
+(
+    "002_entity_embeddings",
+    r#"
+    -- Add embedding column to entities table (episodes already has it from 001)
+    -- We use a Rust-side check since SQLite ALTER TABLE ADD COLUMN is not idempotent
+    "#,
 )];
 
 pub fn run_migrations(conn: &Connection) -> Result<()> {
@@ -162,7 +170,20 @@ pub fn run_migrations(conn: &Connection) -> Result<()> {
         )?;
 
         if !already_applied {
-            conn.execute_batch(sql)?;
+            // Migration 002: add embedding column to entities if not present
+            if *version == "002_entity_embeddings" {
+                let has_col: bool = {
+                    let mut col_stmt = conn.prepare(
+                        "SELECT COUNT(*) FROM pragma_table_info('entities') WHERE name = 'embedding'",
+                    )?;
+                    col_stmt.query_row([], |row| row.get::<_, i64>(0)).map(|n| n > 0)?
+                };
+                if !has_col {
+                    conn.execute_batch("ALTER TABLE entities ADD COLUMN embedding BLOB;")?;
+                }
+            } else {
+                conn.execute_batch(sql)?;
+            }
             conn.execute(
                 "INSERT INTO _migrations (version, applied_at) VALUES (?1, ?2)",
                 rusqlite::params![version, chrono::Utc::now().to_rfc3339()],

crates/ctxgraph-core/src/storage/migrations.rs

@@ -365,6 +365,39 @@ impl Storage {
         Ok(results)
     }
 
+    // ── Embeddings ──
+
+    /// Store a raw f32 embedding blob for an episode.
+    pub fn store_episode_embedding(&self, episode_id: &str, data: &[u8]) -> Result<()> {
+        self.conn.execute(
+            "UPDATE episodes SET embedding = ?1 WHERE id = ?2",
+            params![data, episode_id],
+        )?;
+        Ok(())
+    }
+
+    /// Store a raw f32 embedding blob for an entity.
+    pub fn store_entity_embedding(&self, entity_id: &str, data: &[u8]) -> Result<()> {
+        self.conn.execute(
+            "UPDATE entities SET embedding = ?1 WHERE id = ?2",
+            params![data, entity_id],
+        )?;
+        Ok(())
+    }
+
+    /// Load all episode embeddings as (id, raw bytes) pairs.
+    pub fn get_all_episode_embeddings(&self) -> Result<Vec<(String, Vec<u8>)>> {
+        let mut stmt = self.conn.prepare(
+            "SELECT id, embedding FROM episodes WHERE embedding IS NOT NULL",
+        )?;
+        let rows = stmt
+            .query_map([], |row| {
+                Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
+            })?
+            .collect::<std::result::Result<Vec<_>, _>>()?;
+        Ok(rows)
+    }
+
     // ── Stats ──
 
     pub fn stats(&self) -> Result<GraphStats> {

crates/ctxgraph-core/src/storage/sqlite.rs

@@ -159,6 +159,13 @@ pub struct SearchResult {
     pub score: f64,
 }
 
+/// Per-episode result from fused (RRF) search.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FusedEpisodeResult {
+    pub episode: Episode,
+    pub score: f64,
+}
+
 /// Graph-wide statistics.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct GraphStats {