From 3585376f991c18f2c5431ac605b678c937c3a13c Mon Sep 17 00:00:00 2001 From: yaopengfei Date: Tue, 11 Aug 2026 06:02:16 +0000 Subject: [PATCH 1/2] refactor(memory_engine): make the RRF damping constant configurable RRF_K was a module constant at 60, the value classic RRF uses for TREC-scale runs of ~1000 hits. Skill sources return ~10 hits, where 60 flattens the whole rank ladder to a 15% score spread -- narrower than the 1.0/0.85 gap between source weights. Weight then decides the order outright and a source's internal rank stops mattering: with the shipped settings no EverOS or Hub hit could ever reach the output unless the same skill also surfaced from Local, which contradicts the weighted blend the config docstring describes. Expose it as skill_forge.router.rrf_k and lower the default to 10. That restores the ladder to an 82% spread, so each source's top hit competes on rank, and it keeps the cross-source agreement bonus proportionate: at 60 two rank-10 hits outscored a single rank-1 hit, at 10 they do not. Local's weight tightens to 0.96 so the three sources interleave rather than tier. No retrieval-quality data backs 10 over 60. The change is motivated by the strict-tiering behaviour above, not by a measured improvement. Co-authored-by: Claude (claude-opus-5) --- raven/config/raven.py | 15 ++++++++-- raven/context_engine/factory.py | 1 + raven/memory_engine/skill_forge/fusion.py | 15 ++++++---- raven/memory_engine/skill_forge/router.py | 3 ++ tests/test_config_raven_sections.py | 8 ++++- tests/test_config_update.py | 2 +- tests/test_context_engine_factory.py | 12 ++++++++ tests/test_skill_router_fusion.py | 36 +++++++++++++++++++++-- 8 files changed, 80 insertions(+), 12 deletions(-) diff --git a/raven/config/raven.py b/raven/config/raven.py index 07f160da..1d7a5527 100644 --- a/raven/config/raven.py +++ b/raven/config/raven.py @@ -1173,7 +1173,7 @@ class SkillForgeRouterConfig(_Base): weights: dict[str, float] = Field( default_factory=lambda: { - "local": 1.0, + "local": 0.96, "everos": 0.9, "hub": 0.85, }, @@ -1181,7 +1181,18 @@ class SkillForgeRouterConfig(_Base): """Per-source RRF weight. Higher = more rank mass when the same skill surfaces from multiple sources. Local highest (hand-curated); Hub (the remote marketplace, replaces the retired Mass source) lowest as - imported/unvalidated; Everos in between (task-specific, auto-evolved).""" + imported/unvalidated; Everos in between (task-specific, auto-evolved). + + Only the ratios matter -- scaling all three leaves the order unchanged. + Read them together with ``rrf_k``: the spread has to stay well inside + the rank ladder that ``rrf_k`` produces, or weight silently overrides + rank and each source becomes a strict tier.""" + + rrf_k: int = Field(default=10, ge=1) + """RRF damping constant, mirroring ``skill_forge.fusion.RRF_K``. + Lower = source-internal rank carries more weight relative to + ``weights``; higher = flatter, so cross-source agreement and source + identity dominate.""" over_fetch_factor: int = 2 """Each source is asked for ``top_k * factor`` hits before fusion diff --git a/raven/context_engine/factory.py b/raven/context_engine/factory.py index 7151324f..1d1d843b 100644 --- a/raven/context_engine/factory.py +++ b/raven/context_engine/factory.py @@ -211,6 +211,7 @@ def _build_router( sources=sources, over_fetch_factor=skill_forge_router_config.over_fetch_factor, dedup_by=skill_forge_router_config.dedup_by, + rrf_k=skill_forge_router_config.rrf_k, ) diff --git a/raven/memory_engine/skill_forge/fusion.py b/raven/memory_engine/skill_forge/fusion.py index d10d6ac0..9429a66e 100644 --- a/raven/memory_engine/skill_forge/fusion.py +++ b/raven/memory_engine/skill_forge/fusion.py @@ -37,16 +37,18 @@ from raven.memory_engine.skill_forge.types import RouterHit -# The "60" in classic RRF — dampens rank effects so a #1 at one source -# doesn't always crowd out top-3 from another. Standard value, kept as -# a module constant so the rare experiment that wants to tune it can. -RRF_K: int = 60 +# Classic RRF uses 60, tuned for TREC-scale runs of ~1000 hits. Sources +# here return ~10, where 60 flattens the whole rank ladder to a 15% score +# spread -- narrower than the 1.0/0.85 source-weight gap, so weight alone +# decides the order and rank stops mattering. 10 keeps that ladder at 82%. +RRF_K: int = 10 def rrf_merge_weighted( source_results: list[tuple[str, float, list[RouterHit]]], k: int, dedup_by: str = "name", + rrf_k: int | None = None, ) -> list[RouterHit]: """Fuse per-source ranked lists into one top-K. @@ -60,6 +62,8 @@ def rrf_merge_weighted( sources surfacing a skill with the same display name are one logical skill. Tests pass ``"qualified_id"`` when they want to verify "no dedup happened" on disjoint hits. + rrf_k: RRF damping constant. ``None`` uses :data:`RRF_K`. Not to + be confused with ``k`` above, which caps the output length. Returns: Up to ``k`` :class:`RouterHit` records, ranked by descending @@ -67,6 +71,7 @@ def rrf_merge_weighted( ``contributing_sources`` (list[str], stable-ordered as encountered) added to its ``meta``. """ + damping = RRF_K if rrf_k is None else rrf_k rrf_scores: dict[str, float] = defaultdict(float) best_hit: dict[str, RouterHit] = {} contributing: dict[str, list[str]] = defaultdict(list) @@ -74,7 +79,7 @@ def rrf_merge_weighted( for source_name, weight, hits in source_results: for rank, hit in enumerate(hits, start=1): key = getattr(hit, dedup_by) - rrf_scores[key] += weight / (RRF_K + rank) + rrf_scores[key] += weight / (damping + rank) contributing[key].append(source_name) prev = best_hit.get(key) # Keep the hit with the higher per-source ``score`` as the diff --git a/raven/memory_engine/skill_forge/router.py b/raven/memory_engine/skill_forge/router.py index 92fe17d9..c2fb78f7 100644 --- a/raven/memory_engine/skill_forge/router.py +++ b/raven/memory_engine/skill_forge/router.py @@ -43,6 +43,7 @@ def __init__( *, over_fetch_factor: int = 2, dedup_by: str = "name", + rrf_k: int | None = None, ) -> None: # The list is captured by reference; callers should pass an # already-frozen tuple if they want to forbid mutation. We @@ -52,6 +53,7 @@ def __init__( self._sources = sources self._over_fetch_factor = max(1, over_fetch_factor) self._dedup_by = dedup_by + self._rrf_k = rrf_k async def select( self, @@ -66,6 +68,7 @@ async def select( [(s.name, s.weight, hits) for s, hits in zip(self._sources, per_source)], k=k, dedup_by=self._dedup_by, + rrf_k=self._rrf_k, ) async def _safe_search( diff --git a/tests/test_config_raven_sections.py b/tests/test_config_raven_sections.py index 9d0dfa14..a3f8955f 100644 --- a/tests/test_config_raven_sections.py +++ b/tests/test_config_raven_sections.py @@ -46,10 +46,11 @@ def test_memory_backend_none_disables(self) -> None: def test_skill_router_defaults(self) -> None: c = SkillForgeRouterConfig() assert c.enabled is True - assert c.weights == {"local": 1.0, "everos": 0.9, "hub": 0.85} + assert c.weights == {"local": 0.96, "everos": 0.9, "hub": 0.85} assert c.over_fetch_factor == 2 assert c.dedup_by == "name" assert c.top_k == 5 + assert c.rrf_k == 10 # Hub is the remote source (replaces the retired Mass source); # disabled until an endpoint is set. assert isinstance(c.hub, HubSourceConfig) @@ -58,6 +59,11 @@ def test_skill_router_defaults(self) -> None: assert c.hub.timeout_s == pytest.approx(2.0) assert c.hub.min_safety == pytest.approx(0.7) + def test_rrf_k_accepts_camel_case_and_rejects_zero(self) -> None: + assert SkillForgeRouterConfig(rrfK=30).rrf_k == 30 + with pytest.raises(ValidationError): + SkillForgeRouterConfig(rrf_k=0) + def test_skill_forge_public_defaults(self) -> None: c = SkillForgeConfig() assert c.embedding_model == "default" diff --git a/tests/test_config_update.py b/tests/test_config_update.py index 37fb23a6..1ac7ae5a 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -279,7 +279,7 @@ def test_init_extension_defaults_seeds_safe_subset(cfg_path: Path) -> None: assert data["skillForge"]["enabled"] is True assert data["skillForge"]["everos"] == {"enabled": True} assert data["skillForge"]["router"]["weights"] == { - "local": 1.0, + "local": 0.96, "everos": 0.9, "hub": 0.85, } diff --git a/tests/test_context_engine_factory.py b/tests/test_context_engine_factory.py index db3e77bd..9e973056 100644 --- a/tests/test_context_engine_factory.py +++ b/tests/test_context_engine_factory.py @@ -85,6 +85,7 @@ def _build_engine( memory_config: MemoryConfig | None = None, model: str = "stub", skill_forge_config: SkillForgeConfig | None = None, + rrf_k: int | None = None, ) -> ContextAssembler: builder = ContextBuilder(workspace=tmp_path) engine = build_context_engine( @@ -99,6 +100,7 @@ def _build_engine( memory_config=memory_config or MemoryConfig(), skill_forge_router_config=SkillForgeRouterConfig( hub=HubSourceConfig(endpoint=hub_endpoint), + **({} if rrf_k is None else {"rrf_k": rrf_k}), ), skill_forge_config=skill_forge_config, ) @@ -190,6 +192,16 @@ def test_hub_source_present_when_endpoint_set(self, tmp_path: Path) -> None: types, _ = _router_sources(_build_engine(tmp_path, backend=_FakeBackend(), hub_endpoint="http://hub.test")) assert HubSkillSource in types + def test_rrf_k_forwarded_from_config(self, tmp_path: Path) -> None: + engine = _build_engine(tmp_path, backend=_FakeBackend(), rrf_k=25) + skills = next(b for b in engine._builders if isinstance(b, SkillsSegmentBuilder)) + assert skills._router._rrf_k == 25 + + def test_rrf_k_defaults_to_config_default(self, tmp_path: Path) -> None: + engine = _build_engine(tmp_path, backend=_FakeBackend()) + skills = next(b for b in engine._builders if isinstance(b, SkillsSegmentBuilder)) + assert skills._router._rrf_k == SkillForgeRouterConfig().rrf_k + def test_track_ids_from_memory_config(self, tmp_path: Path) -> None: engine = _build_engine( tmp_path, diff --git a/tests/test_skill_router_fusion.py b/tests/test_skill_router_fusion.py index f91b652c..b2129ad3 100644 --- a/tests/test_skill_router_fusion.py +++ b/tests/test_skill_router_fusion.py @@ -7,6 +7,7 @@ import pytest from raven.memory_engine.skill_forge import ( + RRF_K, RouterHit, SkillForgeRouter, rrf_merge_weighted, @@ -82,8 +83,25 @@ def test_contributing_sources_recorded(self) -> None: def test_rrf_score_written_to_meta(self) -> None: hits = [_hit("a/1", "x")] out = rrf_merge_weighted([("a", 1.0, hits)], k=5) - # RRF score for rank-1 of single source with w=1: 1/(60+1) ~= 0.01639 - assert out[0].meta["rrf_score"] == pytest.approx(1.0 / 61.0) + # RRF score for rank-1 of single source with w=1: 1/(RRF_K+1) + assert out[0].meta["rrf_score"] == pytest.approx(1.0 / (RRF_K + 1)) + + def test_rrf_k_override_steepens_rank_ladder(self) -> None: + """A larger rrf_k flattens rank differences until source weight + alone decides the order; the default keeps rank competitive.""" + local = [ + _hit("local/a", "alpha"), + _hit("local/b", "beta"), + _hit("local/c", "gamma"), + ] + hub = [_hit("hub/d", "delta")] + args = [("local", 1.0, local), ("hub", 0.85, hub)] + + names = [h.name for h in rrf_merge_weighted(args, k=5)] + assert names.index("delta") < names.index("gamma") + + names = [h.name for h in rrf_merge_weighted(args, k=5, rrf_k=60)] + assert names.index("gamma") < names.index("delta") def test_weight_affects_relative_ranking(self) -> None: """A hit ranked #3 in a high-weight source can outrank a hit @@ -101,7 +119,7 @@ def test_weight_affects_relative_ranking(self) -> None: k=5, ) names = [h.name for h in out] - # local rank-3 (1/63 = ~0.0159) > mass rank-1 (0.01/61 = ~0.00016) + # local rank-3 (1/(RRF_K+3)) > mass rank-1 (0.01/(RRF_K+1)), # so 'gamma' must outrank 'delta'. assert names.index("gamma") < names.index("delta") @@ -178,6 +196,18 @@ async def test_fans_out_to_all_sources(self) -> None: names = {h.name for h in out} assert names == {"x", "y"} + async def test_rrf_k_reaches_the_fusion(self) -> None: + """The constructor arg must survive the hop into rrf_merge_weighted. + Asserting the exact score catches a dropped pass-through, which + would otherwise silently fall back to the module default.""" + a = _StubSource("local", 1.0, [_hit("local/x", "x")]) + + out = await SkillForgeRouter([a], rrf_k=60).select("q", history=[], k=5) + assert out[0].meta["rrf_score"] == pytest.approx(1.0 / 61.0) + + out = await SkillForgeRouter([a]).select("q", history=[], k=5) + assert out[0].meta["rrf_score"] == pytest.approx(1.0 / (RRF_K + 1)) + async def test_over_fetches_per_source(self) -> None: """Default over_fetch_factor=2 means each source is asked for k*2.""" a = _StubSource("local", 1.0, [_hit("local/x", "x")]) From 2cc24c3a181cb675c2898b082db9606b92bae02b Mon Sep 17 00:00:00 2001 From: yaopengfei Date: Tue, 11 Aug 2026 08:21:14 +0000 Subject: [PATCH 2/2] docs(memory_engine): correct stale RRF constant references The module docstring in fusion.py still stated the formula with k = 60 and reused the name k for the damping constant, which now collides with the output-cap argument of the same name. CONTEXT.md still described Local's source weight as 1.0. Also assert the ordering the shipped config produces instead of only describing it: the interleave held solely in prose, and the value assertion in test_skill_router_defaults gives no signal that slot order moved. docs/architecture/skill_hub_retrieval.svg carries the same stale formula. Left untouched: repository policy excludes SVG assets from commits. Co-authored-by: Claude (claude-opus-5) --- CONTEXT.md | 2 +- raven/memory_engine/skill_forge/fusion.py | 23 +++++++++++++---------- tests/test_skill_router_fusion.py | 23 +++++++++++++++++++++++ 3 files changed, 37 insertions(+), 11 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 32bedf0d..4d31ae08 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -414,7 +414,7 @@ A remote OpenAPI skill marketplace, configured via `skillForge.router.hub` (`end `api_key` / `timeout_s` / `min_safety`; `endpoint=None` disables it). `SkillHubClient` offers progressive disclosure — `search()` (metadata-only discovery), `get()` (skill body), `install()` (download + safe extract); during routing `HubSkillSource` feeds metadata-only -candidates into the weighted RRF (weight 0.85, below Local 1.0 and Everos 0.9), and the +candidates into the weighted RRF (weight 0.85, below Local 0.96 and Everos 0.9), and the `read_skill` / `use_skill` tools do on-demand body fetch / script materialization. Replaces the retired "Mass" source. diff --git a/raven/memory_engine/skill_forge/fusion.py b/raven/memory_engine/skill_forge/fusion.py index 9429a66e..f9164630 100644 --- a/raven/memory_engine/skill_forge/fusion.py +++ b/raven/memory_engine/skill_forge/fusion.py @@ -1,15 +1,18 @@ """Weighted Reciprocal Rank Fusion across heterogeneous skill sources. -The classic RRF formula sums ``1 / (k + rank_i(d))`` over the sources -that hit document ``d``. Multi-source skill retrieval needs a small -extension: each source carries a **trust weight** so curated content -(Local) outranks imported content (Mass) at equal rank. The weighted -form is:: - - rrf_score(d) = Σ_i w_i / (k + rank_i(d)) - -with ``k = 60`` (the long-standing RRF constant) and the per-source -``w_i`` coming from the source's :attr:`SkillSource.weight` attribute. +The classic RRF formula sums ``1 / (rrf_k + rank_i(d))`` over the +sources that hit document ``d``. Multi-source skill retrieval needs a +small extension: each source carries a **trust weight** so curated +content (Local) outranks imported content (Mass) at equal rank. The +weighted form is:: + + rrf_score(d) = Σ_i w_i / (rrf_k + rank_i(d)) + +with ``rrf_k`` defaulting to :data:`RRF_K` and overridable per call or +via ``skillForge.router.rrfK``, and the per-source ``w_i`` coming from +the source's :attr:`SkillSource.weight` attribute. Note that ``rrf_k`` +is the damping constant, distinct from the ``k`` argument of +:func:`rrf_merge_weighted`, which caps the output length. Three additional behaviors are baked in: diff --git a/tests/test_skill_router_fusion.py b/tests/test_skill_router_fusion.py index b2129ad3..b4af6698 100644 --- a/tests/test_skill_router_fusion.py +++ b/tests/test_skill_router_fusion.py @@ -6,6 +6,7 @@ import pytest +from raven.config.raven import SkillForgeRouterConfig from raven.memory_engine.skill_forge import ( RRF_K, RouterHit, @@ -208,6 +209,28 @@ async def test_rrf_k_reaches_the_fusion(self) -> None: out = await SkillForgeRouter([a]).select("q", history=[], k=5) assert out[0].meta["rrf_score"] == pytest.approx(1.0 / (RRF_K + 1)) + async def test_shipped_defaults_interleave_the_three_sources(self) -> None: + """The ordering the shipped config produces, asserted rather than + described: every source places its top hit and Local keeps slot 1.""" + cfg = SkillForgeRouterConfig() + sources = [ + _StubSource( + name, + cfg.weights[name], + [_hit(f"{name}/1", f"{name}#1"), _hit(f"{name}/2", f"{name}#2")], + ) + for name in ("local", "everos", "hub") + ] + router = SkillForgeRouter(sources, rrf_k=cfg.rrf_k) + out = await router.select("q", history=[], k=5) + assert [h.name for h in out] == [ + "local#1", + "everos#1", + "local#2", + "hub#1", + "everos#2", + ] + async def test_over_fetches_per_source(self) -> None: """Default over_fetch_factor=2 means each source is asked for k*2.""" a = _StubSource("local", 1.0, [_hit("local/x", "x")])