diff --git a/CONTEXT.md b/CONTEXT.md index 32bedf0..4d31ae0 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/config/raven.py b/raven/config/raven.py index 07f160d..1d7a552 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 7151324..1d1d843 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 d10d6ac..f916463 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:: +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 / (k + rank_i(d)) + rrf_score(d) = Σ_i w_i / (rrf_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. +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: @@ -37,16 +40,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 +65,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 +74,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 +82,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 92fe17d..c2fb78f 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 9d0dfa1..a3f8955 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 37fb23a..1ac7ae5 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 db3e77b..9e97305 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 f91b652..b4af669 100644 --- a/tests/test_skill_router_fusion.py +++ b/tests/test_skill_router_fusion.py @@ -6,7 +6,9 @@ import pytest +from raven.config.raven import SkillForgeRouterConfig from raven.memory_engine.skill_forge import ( + RRF_K, RouterHit, SkillForgeRouter, rrf_merge_weighted, @@ -82,8 +84,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 +120,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 +197,40 @@ 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_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")])