From 09852255e29c0fcd9150d70955030cb6d83aac85 Mon Sep 17 00:00:00 2001 From: sabhatinas Date: Fri, 10 Jul 2026 00:37:31 +0000 Subject: [PATCH] feat(stage-router): replace linear clip with tanh sigmoid in scorer --- .../lib/processors/stage_router/scorer.py | 39 ++++++++++++++---- tests/test_stage_router_scorer.py | 40 ++++++++----------- 2 files changed, 47 insertions(+), 32 deletions(-) diff --git a/switchyard/lib/processors/stage_router/scorer.py b/switchyard/lib/processors/stage_router/scorer.py index 0c3763ad..bc439024 100644 --- a/switchyard/lib/processors/stage_router/scorer.py +++ b/switchyard/lib/processors/stage_router/scorer.py @@ -1,17 +1,27 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Weighted linear scorer: signed score in ``[-1, +1]``, confidence = ``abs(score)``.""" +"""Weighted scorer with sigmoid (tanh) shaping: signed score in (-1, +1), confidence = abs(score). + +The raw weighted sum is passed through tanh(k * raw) before thresholding. This amplifies +moderate signals toward ±1, ensuring that efficient turns with several weak negative +dimensions cross the confidence threshold rather than sitting in the ambiguous zone. + +Score zones relative to threshold t: + [-1, -t) → strong efficient signal → route EFFICIENT + [-t, t] → ambiguous → fall through to classifier / default + (t, 1] → strong capable signal → route CAPABLE +""" from __future__ import annotations +import math from collections.abc import Mapping from dataclasses import dataclass, field from switchyard.lib.processors.stage_router.dimensions import CodingAgentDimensions -#: Default linear weights. Positive ⇒ CAPABLE; negative ⇒ EFFICIENT. Calibrated so -#: a single high-impact axis lands past the 0.5 default confidence threshold. +#: Default linear weights. Positive ⇒ CAPABLE; negative ⇒ EFFICIENT. DEFAULT_WEIGHTS: Mapping[str, float] = { "severity": 0.80, "stuck_exploring": 0.70, @@ -25,10 +35,18 @@ "no_error_streak_intensity": -0.20, } +#: Sigmoid steepness. tanh(k * raw): k=2 means raw=±0.5 → score≈±0.76, +#: pushing moderate efficient/capable signals past the default t=0.5 threshold. +DEFAULT_STEEPNESS: float = 2.0 + @dataclass(frozen=True) class ScoreResult: - """Output of :func:`score`. ``confidence == abs(score)`` by construction.""" + """Output of :func:`score`. ``confidence == abs(score)`` by construction. + + ``contributions`` are the raw per-dimension products (before sigmoid); + their sum is the pre-sigmoid input, not necessarily equal to ``score``. + """ score: float confidence: float @@ -39,8 +57,13 @@ def score( dimensions: CodingAgentDimensions, *, weights: Mapping[str, float] = DEFAULT_WEIGHTS, + steepness: float = DEFAULT_STEEPNESS, ) -> ScoreResult: - """Score ``dimensions`` against ``weights``; raw sum is clipped to ``[-1, +1]``.""" + """Score ``dimensions`` against ``weights`` with sigmoid shaping. + + Raw weighted sum is passed through ``tanh(steepness * raw)`` to produce a + score in (-1, +1). Confidence is ``abs(score)``. + """ contributions: dict[str, float] = {} raw = 0.0 for field_name, weight in weights.items(): @@ -48,8 +71,8 @@ def score( contribution = value * weight contributions[field_name] = contribution raw += contribution - clipped = max(-1.0, min(1.0, raw)) - return ScoreResult(score=clipped, confidence=abs(clipped), contributions=contributions) + shaped = math.tanh(steepness * raw) + return ScoreResult(score=shaped, confidence=abs(shaped), contributions=contributions) -__all__ = ["DEFAULT_WEIGHTS", "ScoreResult", "score"] +__all__ = ["DEFAULT_STEEPNESS", "DEFAULT_WEIGHTS", "ScoreResult", "score"] diff --git a/tests/test_stage_router_scorer.py b/tests/test_stage_router_scorer.py index ed2aa6b0..77a7b3fc 100644 --- a/tests/test_stage_router_scorer.py +++ b/tests/test_stage_router_scorer.py @@ -11,7 +11,9 @@ CodingAgentDimensions, from_signal, ) -from switchyard.lib.processors.stage_router.scorer import DEFAULT_WEIGHTS, score +import math + +from switchyard.lib.processors.stage_router.scorer import DEFAULT_STEEPNESS, DEFAULT_WEIGHTS, score from switchyard_rust.components import DimensionCollector from switchyard_rust.core import ChatRequest, ProxyContext @@ -52,17 +54,17 @@ def test_tests_passed_pushes_toward_efficient(): assert result.score < 0 -def test_score_is_clipped_to_unit_interval(): - """Out-of-range weighted sums must be clamped to [-1, 1].""" +def test_score_bounded_by_unit_interval(): + """tanh keeps score strictly in (-1, 1); extreme raw sums saturate near ±1.""" dims = CodingAgentDimensions(**{**_zero_dimensions().__dict__, "severity": 1.0}) - # Force overflow on both sides: a 5.0-magnitude weight × 1.0 dimension - # yields raw ±5.0, which must clip to ±1.0 with confidence 1.0. high = score(dims, weights={"severity": 5.0}) - assert high.score == 1.0 - assert high.confidence == 1.0 + assert -1.0 < high.score <= 1.0 + assert high.score > 0.99 # tanh(10) is ~1.0 to 5 decimal places + assert high.confidence == abs(high.score) low = score(dims, weights={"severity": -5.0}) - assert low.score == -1.0 - assert low.confidence == 1.0 + assert -1.0 <= low.score < 1.0 + assert low.score < -0.99 + assert low.confidence == abs(low.score) def test_custom_weights_can_invert_decision(): @@ -74,23 +76,13 @@ def test_custom_weights_can_invert_decision(): assert inverted.score < 0 -def test_contributions_sum_matches_unclipped_score(): - """When no clipping fires, ``sum(contributions) == score`` exactly.""" +def test_contributions_are_pre_sigmoid_raw_products(): + """contributions are the raw weight×dim products; score = tanh(k * sum(contributions)).""" dims = CodingAgentDimensions(**{**_zero_dimensions().__dict__, "tests_passed": 1.0}) result = score(dims) - expected = sum(result.contributions.values()) - assert abs(expected - result.score) < 1e-9 - - -def test_contributions_can_exceed_clipped_score(): - """When clipping fires, ``sum(contributions)`` may exceed ``|score|``.""" - dims = CodingAgentDimensions(**{**_zero_dimensions().__dict__, "severity": 1.0}) - # Raw = 5.0 → clipped to 1.0; contributions remain unclipped. - result = score(dims, weights={"severity": 5.0}) - raw = sum(result.contributions.values()) - assert raw == 5.0 - assert result.score == 1.0 - assert raw > result.score + raw_sum = sum(result.contributions.values()) + expected_score = math.tanh(DEFAULT_STEEPNESS * raw_sum) + assert abs(result.score - expected_score) < 1e-9 @pytest.mark.asyncio