From d2cb3db07ad54bf404d8357a2b9379eb67f98f72 Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 18 May 2026 19:15:14 +0200 Subject: [PATCH] feat: Mode A hybrid pipeline with graceful degradation to C Co-Authored-By: Claude Opus 4.7 (1M context) --- src/pipeline.rs | 123 ++++++++++++++++++++++++++++++--- tests/pipeline_mode_a_tests.rs | 13 ++++ 2 files changed, 127 insertions(+), 9 deletions(-) create mode 100644 tests/pipeline_mode_a_tests.rs diff --git a/src/pipeline.rs b/src/pipeline.rs index 17c9fc2..42cf1bd 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -1,10 +1,14 @@ use crate::claml; use crate::corpus::{self}; -use crate::fusion::cross_segment_dedupe; +use crate::embed::EmbeddingStore; +use crate::fusion::{cross_segment_dedupe, rrf_merge}; +use crate::ionos::IonosClient; use crate::lexical::LexicalIndex; use crate::model::*; +use crate::rerank; use crate::segment::split_segments; use crate::tags::tags_for; +use crate::vector::VectorIndex; use std::collections::HashMap; use std::time::Instant; @@ -14,6 +18,8 @@ pub struct Pipeline { pub lexical: LexicalIndex, pub pool_size: usize, pub valid_by_code: HashMap, + cfg: Config, + vector: Option, } impl Pipeline { @@ -31,7 +37,26 @@ impl Pipeline { } m }; - Ok(Self { entries, meta, lexical, pool_size: cfg.pool_size, valid_by_code }) + // Build the semantic index only if EVERY corpus entry already has a + // cached embedding; otherwise leave it `None` so Mode A degrades to C + // rather than mispairing partial vectors. + let vector = match EmbeddingStore::open(&cfg.index_dir, &cfg.embed_model) { + Ok(store) if entries.iter().all(|e| store.has(&e.text)) => { + let vecs: Option>> = + entries.iter().map(|e| store.get(&e.text)).collect(); + vecs.map(VectorIndex::from_vectors) + } + _ => None, + }; + Ok(Self { + entries, + meta, + lexical, + pool_size: cfg.pool_size, + valid_by_code, + cfg: cfg.clone(), + vector, + }) } /// Look up ICD metadata: exact code first, then — only for 6-char dotted codes @@ -83,12 +108,10 @@ impl Pipeline { }) } - pub fn suggest(&self, dictation: &str, mode: Mode, filter: &Filter, top_k: usize) - -> SuggestResult { - let t0 = Instant::now(); - let segments = split_segments(dictation); + /// Mode C candidate collection: lexical pool per segment. This is the + /// exact original behavior; it is also the degrade fallback for Mode A. + fn collect_mode_c(&self, segments: &[String]) -> Vec { let mut candidates: Vec = Vec::new(); - for (si, seg) in segments.iter().enumerate() { let lex = self.lexical.search(seg, self.pool_size).unwrap_or_default(); for (entry_idx, score) in lex { @@ -105,6 +128,88 @@ impl Pipeline { } } } + candidates + } + + /// Mode A candidate collection: per segment, fuse a lexical pool with a + /// semantic pool (IONOS embedding + in-RAM vector search) via RRF, then + /// cross-encoder rerank the fused pool. ANY error (IONOS unreachable, no + /// vector index, embed/rerank failure) propagates so `suggest` degrades. + fn collect_mode_a(&self, segments: &[String], calls: &mut usize) + -> Result, AppError> { + let client = IonosClient::new(&self.cfg.ionos_base_url, &self.cfg.token_path)?; + let vector = self.vector.as_ref() + .ok_or_else(|| AppError::Ionos("no vector index".into()))?; + let mut candidates: Vec = Vec::new(); + + for (si, seg) in segments.iter().enumerate() { + // Lexical pool (entry indices). + let lex = self.lexical.search(seg, self.pool_size).unwrap_or_default(); + let lex_ids: Vec = lex.iter().map(|(i, _)| *i).collect(); + + // Semantic pool: embed the segment, then exact cosine search. + let body = serde_json::json!({ + "model": self.cfg.embed_model, "input": [seg] + }); + let resp: serde_json::Value = client.post_json("/embeddings", &body)?; + *calls += 1; + let q: Vec = resp["data"][0]["embedding"].as_array() + .ok_or_else(|| AppError::Ionos("embeddings: no data[0].embedding".into()))? + .iter() + .map(|x| x.as_f64() + .ok_or_else(|| AppError::Ionos("embeddings: non-numeric component".into()))) + .collect::, _>>()? + .into_iter().map(|f| f as f32).collect(); + let sem = vector.top_k(&q, self.pool_size); + let sem_ids: Vec = sem.iter().map(|(i, _)| *i).collect(); + + // Fuse the two ranked index lists; cap at pool_size. + let mut fused = rrf_merge(&lex_ids, &sem_ids, 60); + fused.truncate(self.pool_size); + + // Cross-encoder rerank the fused pool. + let docs: Vec = fused.iter() + .map(|&i| self.entries[i].text.clone()) + .collect(); + let scores = rerank::rerank(&client, &self.cfg.rerank_model, seg, &docs)?; + *calls += 1; + + for (rank, &entry_idx) in fused.iter().enumerate() { + let e = &self.entries[entry_idx]; + if let Some(code) = corpus::primary_code(e) { + candidates.push(Candidate { + icd_code: code.to_string(), + alpha_text: e.text.clone(), + segment_idx: si, + lexical: None, + semantic: None, + rerank: scores.get(rank).copied(), + }); + } + } + } + Ok(candidates) + } + + pub fn suggest(&self, dictation: &str, mode: Mode, filter: &Filter, top_k: usize) + -> SuggestResult { + let t0 = Instant::now(); + let segments = split_segments(dictation); + let mut ionos_calls = 0usize; + let mut degraded = false; + + let candidates: Vec = match mode { + Mode::C => self.collect_mode_c(&segments), + Mode::A => match self.collect_mode_a(&segments, &mut ionos_calls) { + Ok(c) => c, + Err(_) => { + // Total degradation: any failure → Mode C fallback, + // request still answered, never panics. + degraded = true; + self.collect_mode_c(&segments) + } + }, + }; let merged = cross_segment_dedupe(candidates); let mut suggestions = Vec::new(); @@ -123,9 +228,9 @@ impl Pipeline { suggestions, diagnostics: Diagnostics { mode: format!("{:?}", mode), - degraded: false, + degraded, segment_count: segments.len(), - ionos_calls: 0, + ionos_calls, millis: t0.elapsed().as_millis(), }, } diff --git a/tests/pipeline_mode_a_tests.rs b/tests/pipeline_mode_a_tests.rs new file mode 100644 index 0000000..a2559da --- /dev/null +++ b/tests/pipeline_mode_a_tests.rs @@ -0,0 +1,13 @@ +use alpha_id::pipeline::Pipeline; +use alpha_id::model::{Config, Filter, Mode}; + +#[test] +fn mode_a_degrades_to_c_when_ionos_unreachable() { + let mut cfg = Config::load("config/default.toml").unwrap(); + cfg.ionos_base_url = "http://127.0.0.1:1".into(); // refused + let p = Pipeline::load(&cfg).unwrap(); + let res = p.suggest("Diabetes mellitus Typ 2", Mode::A, &Filter::default(), 10); + assert!(res.diagnostics.degraded); // fell back + assert!(!res.suggestions.is_empty()); // still answered + assert!(res.suggestions.iter().any(|s| s.icd_code.starts_with("E11"))); +}