feat(agent): HDF5Memory::search with source filters, re-ranking, confidence

`HDF5Memory::search(query_embedding, query_text, &SearchOptions)` is the
store's full search path. `SearchOptions::new(k)` is plain hybrid search
with the tuned default fusion; each further stage is opt-in:

- `with_sources([..])`: only records from these source channels. The
  filter applies before ranking, so a filtered search still returns up
  to k results, normalised over what it can return. The HNSW pool is
  over-fetched in proportion to what the filter removes, and the allowed
  records are scanned exactly whenever that costs fewer distance
  evaluations than the index would (~pool x M) — and as the fallback if
  the pool comes back short. Keyword matches are filtered too.
- `with_rerank(ReRankConfig)` re-ranks a max(3k, 10) candidate pool by
  relevance, recency, source authority and activation;
  `with_confidence(ConfidenceConfig)` drops low-confidence results;
  `at_time(now)` pins the recency clock.

These were reachable only through the OpenClaw backend, which is now
`search` with both on. Its Hebbian boost now goes to the k results it
returns rather than the whole 3k candidate pool. `hybrid_search` and
`hybrid_search_with` are wrappers and unchanged (tested bit for bit).

Measured on tank (search_harness --options-study --full, 3 runs): at
100K every filter — 50%, 10%, 1% of the store, and records far from the
query — returns the exact filtered top 10, and none is slower than an
unfiltered search (1%: 2.3 ms vs 4.6 ms). Re-rank + confidence costs
about 3%. A first version decided between index and exact scan by pool
size vs store size; it measured 0.976 recall at 12.3 ms on the
far-from-query filter, which is why the rule compares costs instead.

Tests: tests/search_options.rs (filter correctness and full pages via
both paths, far-from-query fallback, edge cases, equality with
hybrid_search_with, re-rank recency, confidence, boost scope).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
osobh
2026-09-24 16:35:42 -05:00
co-authored by Claude Opus 5.5
parent d0db83812b
commit c470244a6f
10 changed files with 951 additions and 110 deletions
+23 -20
View File
@@ -65,31 +65,34 @@ pub fn hybrid_search_fused(
) -> Vec<(usize, f32)> {
// Get raw scores from both systems. Request all results so normalization
// covers the full distribution.
// Use parallel search when rayon feature is enabled and vector count > 10K.
let vec_scores = {
#[cfg(feature = "parallel")]
{
if vectors.count() > 10_000 {
vector_search::parallel_cosine_batch(
query_embedding,
vectors,
tombstones,
vectors.count(),
)
} else {
vector_search::cosine_similarity_batch(query_embedding, vectors, tombstones)
}
}
#[cfg(not(feature = "parallel"))]
{
vector_search::cosine_similarity_batch(query_embedding, vectors, tombstones)
}
};
let vec_scores = exact_vector_scores(query_embedding, vectors, tombstones);
let kw_scores = bm25_index.scores(query_text);
fuse(vec_scores, kw_scores, fusion, k)
}
/// Cosine similarity of `query_embedding` to every vector whose `skip` byte is
/// 0 (a tombstone, or any other exclusion mask). Parallel above 10K vectors
/// when the `parallel` feature is on.
pub fn exact_vector_scores(
query_embedding: &[f32],
vectors: &(impl crate::vector_search::VectorSet + Sync + ?Sized),
skip: &[u8],
) -> Vec<(usize, f32)> {
#[cfg(feature = "parallel")]
{
if vectors.count() > 10_000 {
return vector_search::parallel_cosine_batch(
query_embedding,
vectors,
skip,
vectors.count(),
);
}
}
vector_search::cosine_similarity_batch(query_embedding, vectors, skip)
}
/// Merge pre-computed vector-similarity and keyword scores into a single ranking.
///
/// Both score sets are independently min-max normalized to [0, 1] and combined
+1
View File
@@ -72,6 +72,7 @@ use ephemeral::{EphemeralConfig, EphemeralStore};
pub use ephemeral::{EphemeralEntry, EphemeralStats};
use knowledge::KnowledgeCache;
use memory_strategy::{Exchange, MemoryStrategy, StrategyOutput};
pub use search::SearchOptions;
use session::SessionCache;
// --- Error type ---
+16 -61
View File
@@ -13,9 +13,8 @@ use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use crate::{
AgentMemory, HDF5Memory, MemoryConfig, MemoryEntry,
confidence::{ConfidenceConfig, ScoredResult, reject_low_confidence},
reranker::{ReRankConfig, RerankInput, rerank},
AgentMemory, HDF5Memory, MemoryConfig, MemoryEntry, SearchOptions,
confidence::ConfidenceConfig, reranker::ReRankConfig,
};
// ─────────────────────────────────────────────────────────────────────────────
@@ -524,71 +523,27 @@ impl ClawhdfBackend {
impl MemoryBackend for ClawhdfBackend {
/// Search using hybrid vector + BM25 retrieval, then re-rank and
/// confidence-filter.
/// confidence-filter — [`HDF5Memory::search`] with both stages on.
fn search(
&mut self,
query_text: &str,
query_embedding: &[f32],
k: usize,
) -> Vec<MemorySearchResult> {
// 1. Hybrid retrieval (vector + BM25, fused by score).
let candidates = k.saturating_mul(3).max(10);
let raw = self.memory.hybrid_search_with(
query_embedding,
query_text,
crate::hybrid::DEFAULT_FUSION,
candidates,
);
if raw.is_empty() {
return Vec::new();
}
let now = Self::now_secs();
// 2. Re-rank using temporal recency, source authority, Hebbian weight.
let rerank_inputs: Vec<RerankInput> = raw
.iter()
.map(|r| RerankInput {
index: r.index,
timestamp: r.timestamp,
source_channel: r.source_channel.clone(),
raw_activation: r.activation,
relevance: r.score,
})
.collect();
let reranked = rerank(&rerank_inputs, &self.rerank_config, now);
// 3. Confidence rejection.
let scored: Vec<ScoredResult> = reranked
.iter()
.map(|r| ScoredResult {
index: r.index,
score: r.combined_score,
})
.collect();
let confident = reject_low_confidence(&scored, &self.confidence_config);
// 4. Map back to MemorySearchResult; preserve raw text via index lookup.
let raw_by_idx: HashMap<usize, &crate::SearchResult> =
raw.iter().map(|r| (r.index, r)).collect();
confident
let options = SearchOptions::new(k)
.with_rerank(self.rerank_config)
.with_confidence(self.confidence_config.clone())
.at_time(Self::now_secs());
self.memory
.search(query_embedding, query_text, &options)
.into_iter()
.take(k)
.filter_map(|sr| {
let r = raw_by_idx.get(&sr.index)?;
let path = r.source_channel.clone();
Some(MemorySearchResult {
text: r.chunk.clone(),
score: sr.score,
path: path.clone(),
line_range: None,
timestamp: Some(r.timestamp),
source: path,
})
.map(|r| MemorySearchResult {
text: r.chunk,
score: r.score,
path: r.source_channel.clone(),
line_range: None,
timestamp: Some(r.timestamp),
source: r.source_channel,
})
.collect()
}
+274 -24
View File
@@ -2,18 +2,107 @@
use std::path::Path;
use std::collections::HashSet;
use crate::bm25;
use crate::confidence::{ConfidenceConfig, ScoredResult, reject_low_confidence};
use crate::hybrid;
use crate::reranker::{ReRankConfig, RerankInput, rerank};
use crate::{HDF5Memory, MAX_ACTIVATION_WEIGHT, MemoryError, Result, SearchResult};
/// Options for [`HDF5Memory::search`].
///
/// [`SearchOptions::new`] is plain hybrid search with the tuned default
/// fusion — the same as `hybrid_search_with(.., hybrid::DEFAULT_FUSION, k)`.
/// Every stage beyond that is opt-in.
#[derive(Debug, Clone)]
pub struct SearchOptions {
/// Number of results to return.
pub k: usize,
/// How the vector and keyword stages are combined.
pub fusion: hybrid::Fusion,
/// Only consider records whose `source_channel` is one of these. The
/// filter applies *before* ranking, so a filtered search still returns up
/// to `k` results and scores are normalised over the records it can
/// return. `None` searches everything; an empty list matches nothing.
pub source_channels: Option<Vec<String>>,
/// Re-rank a candidate pool by retrieval relevance, recency, source
/// authority and activation — the pipeline the OpenClaw backend runs.
pub rerank: Option<ReRankConfig>,
/// Candidates retrieved for re-ranking; 0 means `max(3k, 10)`.
pub rerank_pool: usize,
/// Drop low-confidence results (after re-ranking, when that is on).
pub confidence: Option<ConfidenceConfig>,
/// The time recency is measured from, in seconds since the epoch.
/// `None` uses the system clock.
pub now: Option<f64>,
}
impl SearchOptions {
pub fn new(k: usize) -> Self {
Self {
k,
fusion: hybrid::DEFAULT_FUSION,
source_channels: None,
rerank: None,
rerank_pool: 0,
confidence: None,
now: None,
}
}
pub fn with_fusion(mut self, fusion: hybrid::Fusion) -> Self {
self.fusion = fusion;
self
}
/// Search only records from these source channels.
pub fn with_sources<S: Into<String>>(mut self, channels: impl IntoIterator<Item = S>) -> Self {
self.source_channels = Some(channels.into_iter().map(Into::into).collect());
self
}
pub fn with_rerank(mut self, config: ReRankConfig) -> Self {
self.rerank = Some(config);
self
}
pub fn with_confidence(mut self, config: ConfidenceConfig) -> Self {
self.confidence = Some(config);
self
}
/// Measure recency from `now` (seconds since the epoch) instead of the
/// system clock — for reproducible results and tests.
pub fn at_time(mut self, now: f64) -> Self {
self.now = Some(now);
self
}
}
impl Default for SearchOptions {
fn default() -> Self {
Self::new(10)
}
}
impl HDF5Memory {
/// Vector + keyword scoring stage of [`HDF5Memory::hybrid_search`].
/// Vector + keyword scoring stage of [`HDF5Memory::search`].
///
/// Without the `hnsw` feature this is a full linear cosine scan (the exact
/// previous behaviour, also used as the correctness oracle in tests). With
/// `hnsw` enabled and an index available, the vector candidates come from an
/// approximate-nearest-neighbour search over an over-fetched pool, then merge
/// with BM25 via the shared [`hybrid::merge_vector_keyword`].
///
/// `exclude`, when given, marks records that must not be returned (1 =
/// excluded; it covers tombstones too). The index is over-fetched in
/// proportion to how much the mask removes. Surfacing `pool` candidates
/// costs the index roughly `pool × M` distance evaluations, while an exact
/// scan of the allowed records costs one each — so whenever that scan is
/// the cheaper of the two it is used instead, and it is also the fallback
/// if the pool comes back with too few allowed hits (the allowed records
/// sit away from the query). A filtered search never comes back short.
#[cfg(feature = "hnsw")]
fn vector_keyword_search(
&mut self,
@@ -22,16 +111,30 @@ impl HDF5Memory {
bm25: &bm25::BM25Index,
fusion: hybrid::Fusion,
k: usize,
exclude: Option<&[u8]>,
) -> Vec<(usize, f32)> {
self.ensure_hnsw_fresh();
let n = self.cache.len();
// Over-fetch so the merge sees a useful vector pool. `ef` is
// configurable, but the pool the fusion stage sees is not tied to it:
// a caller lowering `ef` for speed should not silently narrow what
// fusion has to work with.
let mut pool = (k * 8).max(64);
let mut allowed = n;
if let Some(ex) = exclude {
allowed = ex.iter().filter(|&&e| e == 0).count();
if allowed == 0 {
return Vec::new();
}
// Expect `pool` allowed hits if the filter is independent of the
// query's neighbourhood.
pool = pool.saturating_mul(n).div_ceil(allowed);
if allowed <= pool.saturating_mul(self.hnsw_m()) {
return self.exact_masked_search(query_embedding, query_text, bm25, fusion, k, ex);
}
}
match self.hnsw.as_ref() {
Some(index) if !index.is_empty() && index.dimension() == query_embedding.len() => {
// Over-fetch so the merge sees a useful vector pool; cosine
// distance from the index converts back to similarity (1 - d).
// `ef` is configurable, but the pool the fusion stage sees is
// not tied to it: a caller lowering `ef` for speed should not
// silently narrow what fusion has to work with.
let pool = (k * 8).max(64);
let ef = self.hnsw_ef_search(k).max(pool);
let candidates = index.search(query_embedding, pool, ef);
// A quantised index returns approximate distances, and no
@@ -42,6 +145,7 @@ impl HDF5Memory {
let exact = index.storage() == clawhdf5_ann::Storage::Int8;
let vec_scores: Vec<(usize, f32)> = candidates
.into_iter()
.filter(|(id, _)| exclude.is_none_or(|ex| ex[*id] == 0))
.map(|(id, dist)| {
let score = if exact {
crate::vector_search::cosine_similarity(
@@ -56,10 +160,54 @@ impl HDF5Memory {
.collect();
// Fusion normalises over every keyword match, so it needs all
// the scores — but not ranked.
let kw_scores = bm25.scores(query_text);
let mut kw_scores = bm25.scores(query_text);
if let Some(ex) = exclude {
if vec_scores.len() < k.min(allowed) {
// The allowed records are not where the index looked.
return self.exact_masked_search(
query_embedding,
query_text,
bm25,
fusion,
k,
ex,
);
}
kw_scores.retain(|(id, _)| ex[*id] == 0);
}
hybrid::fuse(vec_scores, kw_scores, fusion, k)
}
_ => hybrid::hybrid_search_fused(
_ => match exclude {
Some(ex) => {
self.exact_masked_search(query_embedding, query_text, bm25, fusion, k, ex)
}
None => hybrid::hybrid_search_fused(
query_embedding,
query_text,
&self.cache.embeddings,
&self.cache.chunks,
&self.cache.tombstones,
bm25,
fusion,
k,
),
},
}
}
#[cfg(not(feature = "hnsw"))]
fn vector_keyword_search(
&mut self,
query_embedding: &[f32],
query_text: &str,
bm25: &bm25::BM25Index,
fusion: hybrid::Fusion,
k: usize,
exclude: Option<&[u8]>,
) -> Vec<(usize, f32)> {
match exclude {
Some(ex) => self.exact_masked_search(query_embedding, query_text, bm25, fusion, k, ex),
None => hybrid::hybrid_search_fused(
query_embedding,
query_text,
&self.cache.embeddings,
@@ -72,25 +220,33 @@ impl HDF5Memory {
}
}
#[cfg(not(feature = "hnsw"))]
fn vector_keyword_search(
&mut self,
/// Exact hybrid search over the records `exclude` leaves (0 = allowed).
fn exact_masked_search(
&self,
query_embedding: &[f32],
query_text: &str,
bm25: &bm25::BM25Index,
fusion: hybrid::Fusion,
k: usize,
exclude: &[u8],
) -> Vec<(usize, f32)> {
hybrid::hybrid_search_fused(
query_embedding,
query_text,
&self.cache.embeddings,
&self.cache.chunks,
&self.cache.tombstones,
bm25,
fusion,
k,
)
let vec_scores =
hybrid::exact_vector_scores(query_embedding, &self.cache.embeddings, exclude);
let mut kw_scores = bm25.scores(query_text);
kw_scores.retain(|(id, _)| exclude.get(*id) == Some(&0));
hybrid::fuse(vec_scores, kw_scores, fusion, k)
}
/// The exclusion mask for a source-channel filter: 1 for a tombstoned
/// record or one from a channel not in `channels`.
fn source_mask(&self, channels: &[String]) -> Vec<u8> {
let allowed: HashSet<&str> = channels.iter().map(String::as_str).collect();
self.cache
.source_channels
.iter()
.zip(&self.cache.tombstones)
.map(|(ch, &t)| u8::from(t != 0 || !allowed.contains(ch.as_str())))
.collect()
}
/// Perform hybrid search combining cosine vector similarity and BM25 keyword search.
@@ -125,12 +281,53 @@ impl HDF5Memory {
fusion: hybrid::Fusion,
k: usize,
) -> Vec<SearchResult> {
self.search(
query_embedding,
query_text,
&SearchOptions::new(k).with_fusion(fusion),
)
}
/// Hybrid search with optional source filtering, re-ranking and
/// confidence rejection — see [`SearchOptions`].
///
/// Stages, in order: vector + keyword retrieval over the records the
/// source filter allows; fusion; scaling by Hebbian activation; re-ranking
/// (if on) of a `rerank_pool` of candidates; confidence rejection (if on);
/// the top `k`. The records returned with a positive score get their
/// Hebbian boost.
pub fn search(
&mut self,
query_embedding: &[f32],
query_text: &str,
options: &SearchOptions,
) -> Vec<SearchResult> {
let k = options.k;
let fetch = match options.rerank {
Some(_) if options.rerank_pool > 0 => options.rerank_pool.max(k),
Some(_) => k.saturating_mul(3).max(10),
None => k,
};
let exclude = options
.source_channels
.as_deref()
.map(|channels| self.source_mask(channels));
// The keyword index lives for the life of the store and is updated
// incrementally. Take it out for the duration of the call so the
// vector stage can borrow `self` mutably, then put it back.
self.ensure_bm25_fresh();
let bm25 = self.bm25.take().expect("ensure_bm25_fresh leaves an index");
let scored = self.vector_keyword_search(query_embedding, query_text, &bm25, fusion, k);
let scored = self.vector_keyword_search(
query_embedding,
query_text,
&bm25,
options.fusion,
fetch,
exclude.as_deref(),
);
self.bm25 = Some(bm25);
let mut results: Vec<SearchResult> = scored
.into_iter()
.map(|(idx, score)| {
@@ -154,6 +351,25 @@ impl HDF5Memory {
.then(a.index.cmp(&b.index))
});
if let Some(config) = &options.rerank {
results = Self::rerank_results(results, config, options.now);
}
if let Some(config) = &options.confidence {
let scored: Vec<ScoredResult> = results
.iter()
.map(|r| ScoredResult {
index: r.index,
score: r.score,
})
.collect();
let keep: HashSet<usize> = reject_low_confidence(&scored, config)
.into_iter()
.map(|r| r.index)
.collect();
results.retain(|r| keep.contains(&r.index));
}
results.truncate(k);
// Only reinforce records that actually matched. When fewer than `k`
// records are relevant, the rest of the list is zero-score filler;
// boosting it would teach the store that arbitrary records are
@@ -164,11 +380,45 @@ impl HDF5Memory {
.map(|r| r.index)
.collect();
self.apply_hebbian_boost(&hit_indices);
self.bm25 = Some(bm25);
results
}
/// Reorder by the re-ranker's combined score, which also becomes each
/// result's `score`.
fn rerank_results(
results: Vec<SearchResult>,
config: &ReRankConfig,
now: Option<f64>,
) -> Vec<SearchResult> {
let now = now.unwrap_or_else(|| {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs_f64())
.unwrap_or(0.0)
});
let inputs: Vec<RerankInput> = results
.iter()
.map(|r| RerankInput {
index: r.index,
timestamp: r.timestamp,
source_channel: r.source_channel.clone(),
raw_activation: r.activation,
relevance: r.score,
})
.collect();
let mut by_index: std::collections::HashMap<usize, SearchResult> =
results.into_iter().map(|r| (r.index, r)).collect();
rerank(&inputs, config, now)
.into_iter()
.filter_map(|rr| {
let mut r = by_index.remove(&rr.index)?;
r.score = rr.combined_score;
Some(r)
})
.collect()
}
/// Reinforce the records a query returned. The new weights are persisted by
/// the next checkpoint (any write that flushes, `flush_wal`, or drop) — not
/// by rewriting the whole store inside the query, which is what made