Merge pull request #105 from kalibr-ai/claude/add-kalibr-auto-scoring-dmoZp

Add continuous scoring and default heuristic scoring to Router

Author: devonakelley

kalibr/router.py

@@ -50,16 +50,25 @@ class Router:
     """
     Routes LLM requests to the best model based on learned outcomes.
 
-    Example:
+    Three scoring modes (in priority order):
+    1. score_when: Continuous scoring (0.0-1.0). Best for quality optimization.
+       Example: score_when=lambda out: min(1.0, len(out) / 500)
+    2. success_when: Binary scoring (True/False). Good for pass/fail checks.
+       Example: success_when=lambda out: "@" in out
+    3. Default: When neither is provided, Kalibr auto-scores using heuristics
+       (response length, structure, finish reason). Gives day-one metrics
+       without any evaluation code.
+
+    Examples:
+        # Continuous scoring (best quality signal)
         router = Router(
             goal="summarize",
             paths=["gpt-4o", "claude-sonnet-4-20250514"],
-            success_when=lambda out: len(out) > 100
+            score_when=lambda out: min(1.0, len(out) / 500)
         )
         response = router.completion(messages=[...])
 
-    Examples:
-        # Simple auto-reporting
+        # Binary auto-reporting
         router = Router(
             goal="extract_email",
             paths=["gpt-4o", "claude-sonnet-4-20250514"],
@@ -68,6 +77,14 @@ class Router:
         response = router.completion(messages=[...])
         # report() called automatically
 
+        # Zero-config (default heuristic scoring)
+        router = Router(
+            goal="chat",
+            paths=["gpt-4o", "claude-sonnet-4-20250514"]
+        )
+        response = router.completion(messages=[...])
+        # Auto-scored using heuristics - no evaluation code needed
+
         # Manual reporting for complex validation
         router = Router(
             goal="book_meeting",
@@ -88,12 +105,22 @@ def __init__(
         goal: str,
         paths: Optional[List[PathSpec]] = None,
         success_when: Optional[Callable[[str], bool]] = None,
+        score_when: Optional[Callable[[str], float]] = None,
         exploration_rate: Optional[float] = None,
         auto_register: bool = True,
     ):
         """
         Initialize router.
 
+        Three scoring modes (in priority order):
+        1. score_when: Continuous scoring (0.0-1.0). Best for quality optimization.
+           Example: score_when=lambda out: min(1.0, len(out) / 500)
+        2. success_when: Binary scoring (True/False). Good for pass/fail checks.
+           Example: success_when=lambda out: "@" in out
+        3. Default: When neither is provided, Kalibr auto-scores using heuristics
+           (response length, structure, finish reason). Gives day-one metrics
+           without any evaluation code.
+
         Args:
             goal: Name of the goal (e.g., "book_meeting", "summarize")
             paths: List of models or path configs. Examples:
@@ -109,6 +136,11 @@ def __init__(
                          Examples:
                              success_when=lambda out: len(out) > 0  # Not empty
                              success_when=lambda out: "@" in out     # Contains email
+            score_when: Optional function to auto-evaluate quality from LLM output.
+                       Takes the output string and returns a float (0.0-1.0).
+                       Takes priority over success_when if both are provided.
+                       Examples:
+                           score_when=lambda out: min(1.0, len(out) / 500)
             exploration_rate: Override exploration rate (0.0-1.0)
             auto_register: If True, register paths on init
         """
@@ -130,6 +162,7 @@ def __init__(
             )
 
         self.success_when = success_when
+        self.score_when = score_when
         self.exploration_rate = exploration_rate
         self._last_trace_id: Optional[str] = None
         self._last_model_id: Optional[str] = None
@@ -293,12 +326,24 @@ def completion(
                     # Success! Update state to reflect which model succeeded
                     self._last_model_id = candidate_model
 
-                    # Auto-report success if success_when provided
-                    if self.success_when and not self._outcome_reported:
+                    # Auto-report if any scoring mechanism is provided (or use defaults)
+                    if not self._outcome_reported:
                         try:
                             output = response.choices[0].message.content or ""
-                            success = self.success_when(output)
-                            self.report(success=success)
+
+                            if self.score_when:
+                                # Priority 1: User-provided continuous scorer
+                                score = self.score_when(output)
+                                score = min(1.0, max(0.0, float(score)))
+                                self.report(success=score >= 0.5, score=score)
+                            elif self.success_when:
+                                # Priority 2: User-provided binary scorer
+                                success = self.success_when(output)
+                                self.report(success=success)
+                            else:
+                                # Priority 3: Default heuristic scoring (zero-config)
+                                score = self._default_score(response)
+                                self.report(success=score >= 0.5, score=score)
                         except Exception as e:
                             logger.warning(f"Auto-outcome evaluation failed: {e}")
 
@@ -336,6 +381,72 @@ def completion(
     # Alias for common naming confusion
     complete = completion
 
+    def _default_score(self, response) -> float:
+        """
+        Compute a heuristic quality score from an LLM response.
+        Used when no success_when or score_when is provided.
+        Gives users day-one quality metrics without writing evaluation code.
+
+        Signals (all normalized to 0-1, then weighted average):
+        - non_empty: 1.0 if response has content, 0.0 if empty
+        - length_score: normalized response length (sigmoid around 200 chars)
+        - structure_score: bonus for JSON validity, markdown headers, bullet points
+        - finish_reason_score: 1.0 for "stop", 0.5 for "length" (truncated), 0.0 for error
+        """
+        try:
+            content = response.choices[0].message.content or ""
+        except (AttributeError, IndexError):
+            return 0.0
+
+        # Signal 1: Non-empty (binary)
+        non_empty = 1.0 if len(content.strip()) > 0 else 0.0
+        if non_empty == 0.0:
+            return 0.0  # Empty response is always 0
+
+        # Signal 2: Response length (sigmoid - most responses 50-2000 chars)
+        import math
+        char_count = len(content)
+        # Sigmoid centered at 200 chars, gives ~0.5 at 200, ~0.95 at 1000
+        length_score = 1.0 / (1.0 + math.exp(-0.005 * (char_count - 200)))
+
+        # Signal 3: Structure (JSON, markdown, lists indicate structured output)
+        structure_score = 0.5  # baseline
+        content_stripped = content.strip()
+        # JSON detection
+        if (content_stripped.startswith('{') and content_stripped.endswith('}')) or \
+           (content_stripped.startswith('[') and content_stripped.endswith(']')):
+            try:
+                import json
+                json.loads(content_stripped)
+                structure_score = 1.0  # Valid JSON
+            except (json.JSONDecodeError, ValueError):
+                structure_score = 0.3  # Looks like JSON but invalid
+        # Markdown/list detection
+        elif any(marker in content for marker in ['## ', '- ', '* ', '1. ', '```']):
+            structure_score = 0.8
+
+        # Signal 4: Finish reason
+        try:
+            finish_reason = response.choices[0].finish_reason
+            if finish_reason == "stop":
+                finish_score = 1.0
+            elif finish_reason == "length":
+                finish_score = 0.5  # Truncated
+            else:
+                finish_score = 0.3  # Unknown/error
+        except (AttributeError, IndexError):
+            finish_score = 0.5
+
+        # Weighted average
+        score = (
+            non_empty * 0.1 +
+            length_score * 0.3 +
+            structure_score * 0.3 +
+            finish_score * 0.3
+        )
+
+        return round(min(1.0, max(0.0, score)), 3)
+
     def report(
         self,
         success: bool,

tests/test_router.py

@@ -1,11 +1,24 @@
 """Tests for Router class."""
 
 import pytest
+from types import SimpleNamespace
 from unittest.mock import patch, MagicMock
 
 from kalibr.router import Router
 
 
+def _make_response(content="Hello world", finish_reason="stop"):
+    """Helper to create a mock OpenAI-format response."""
+    return SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(content=content),
+                finish_reason=finish_reason,
+            )
+        ]
+    )
+
+
 class TestRouterInit:
     def test_basic_init(self):
         router = Router(goal="test", paths=["gpt-4o"], auto_register=False)
@@ -54,3 +67,168 @@ def test_double_report_warning(self):
         router._outcome_reported = True
         # Should not raise, just warn
         router.report(success=True)
+
+
+class TestDefaultScore:
+    def test_empty_response_returns_zero(self):
+        router = Router(goal="test", auto_register=False)
+        response = _make_response(content="")
+        assert router._default_score(response) == 0.0
+
+    def test_none_content_returns_zero(self):
+        router = Router(goal="test", auto_register=False)
+        response = _make_response(content=None)
+        assert router._default_score(response) == 0.0
+
+    def test_whitespace_only_returns_zero(self):
+        router = Router(goal="test", auto_register=False)
+        response = _make_response(content="   \n  ")
+        assert router._default_score(response) == 0.0
+
+    def test_normal_text_above_half(self):
+        router = Router(goal="test", auto_register=False)
+        response = _make_response(content="This is a normal response with enough text to be useful." * 5)
+        score = router._default_score(response)
+        assert score > 0.5
+
+    def test_valid_json_scores_high(self):
+        router = Router(goal="test", auto_register=False)
+        response = _make_response(content='{"name": "Alice", "email": "[email protected]", "age": 30}')
+        score = router._default_score(response)
+        # Valid JSON should get structure_score=1.0
+        assert score > 0.5
+
+    def test_invalid_json_scores_lower(self):
+        router = Router(goal="test", auto_register=False)
+        valid_response = _make_response(content='{"name": "Alice", "email": "[email protected]"}')
+        invalid_response = _make_response(content='{name: Alice, email: broken}')
+        valid_score = router._default_score(valid_response)
+        invalid_score = router._default_score(invalid_response)
+        assert valid_score > invalid_score
+
+    def test_markdown_gets_structure_bonus(self):
+        router = Router(goal="test", auto_register=False)
+        plain = _make_response(content="Just some plain text response here.")
+        markdown = _make_response(content="## Header\n- Item 1\n- Item 2\n- Item 3")
+        plain_score = router._default_score(plain)
+        md_score = router._default_score(markdown)
+        assert md_score > plain_score
+
+    def test_truncated_response_scores_lower(self):
+        router = Router(goal="test", auto_register=False)
+        text = "A decent response." * 20
+        stopped = _make_response(content=text, finish_reason="stop")
+        truncated = _make_response(content=text, finish_reason="length")
+        assert router._default_score(stopped) > router._default_score(truncated)
+
+    def test_score_between_zero_and_one(self):
+        router = Router(goal="test", auto_register=False)
+        for content in ["x", "hello world", "a" * 5000, '{"key": "value"}']:
+            score = router._default_score(_make_response(content=content))
+            assert 0.0 <= score <= 1.0
+
+    def test_malformed_response_returns_zero(self):
+        router = Router(goal="test", auto_register=False)
+        # Response with no choices attribute
+        assert router._default_score(SimpleNamespace()) == 0.0
+        # Response with empty choices
+        assert router._default_score(SimpleNamespace(choices=[])) == 0.0
+
+
+class TestScoreWhen:
+    def test_score_when_stored(self):
+        scorer = lambda out: 0.8
+        router = Router(goal="test", score_when=scorer, auto_register=False)
+        assert router.score_when is scorer
+
+    @patch("kalibr.router.Router._dispatch")
+    @patch("kalibr.router.Router.report")
+    @patch("kalibr.intelligence.decide")
+    def test_score_when_called_with_output(self, mock_decide, mock_report, mock_dispatch):
+        mock_decide.return_value = {"model_id": "gpt-4o", "trace_id": "abc123"}
+        mock_dispatch.return_value = _make_response(content="test output")
+
+        router = Router(goal="test", score_when=lambda out: 0.75, auto_register=False)
+        router.completion(messages=[{"role": "user", "content": "hi"}])
+
+        mock_report.assert_called_once_with(success=True, score=0.75)
+
+    @patch("kalibr.router.Router._dispatch")
+    @patch("kalibr.router.Router.report")
+    @patch("kalibr.intelligence.decide")
+    def test_score_when_clamps_values(self, mock_decide, mock_report, mock_dispatch):
+        mock_decide.return_value = {"model_id": "gpt-4o", "trace_id": "abc123"}
+        mock_dispatch.return_value = _make_response(content="test")
+
+        # score_when returns value > 1.0, should be clamped
+        router = Router(goal="test", score_when=lambda out: 1.5, auto_register=False)
+        router.completion(messages=[{"role": "user", "content": "hi"}])
+
+        mock_report.assert_called_once_with(success=True, score=1.0)
+
+    @patch("kalibr.router.Router._dispatch")
+    @patch("kalibr.router.Router.report")
+    @patch("kalibr.intelligence.decide")
+    def test_score_when_low_score_reports_failure(self, mock_decide, mock_report, mock_dispatch):
+        mock_decide.return_value = {"model_id": "gpt-4o", "trace_id": "abc123"}
+        mock_dispatch.return_value = _make_response(content="test")
+
+        router = Router(goal="test", score_when=lambda out: 0.2, auto_register=False)
+        router.completion(messages=[{"role": "user", "content": "hi"}])
+
+        mock_report.assert_called_once_with(success=False, score=0.2)
+
+    @patch("kalibr.router.Router._dispatch")
+    @patch("kalibr.router.Router.report")
+    @patch("kalibr.intelligence.decide")
+    def test_score_when_takes_priority_over_success_when(self, mock_decide, mock_report, mock_dispatch):
+        mock_decide.return_value = {"model_id": "gpt-4o", "trace_id": "abc123"}
+        mock_dispatch.return_value = _make_response(content="test output")
+
+        # Both provided - score_when should win
+        router = Router(
+            goal="test",
+            score_when=lambda out: 0.9,
+            success_when=lambda out: False,  # Would report failure if used
+            auto_register=False,
+        )
+        router.completion(messages=[{"role": "user", "content": "hi"}])
+
+        # score_when used (score=0.9 -> success=True), not success_when (which would give False)
+        mock_report.assert_called_once_with(success=True, score=0.9)
+
+
+class TestDefaultScoringIntegration:
+    @patch("kalibr.router.Router._dispatch")
+    @patch("kalibr.router.Router.report")
+    @patch("kalibr.intelligence.decide")
+    def test_default_scoring_fires_when_no_scorers(self, mock_decide, mock_report, mock_dispatch):
+        mock_decide.return_value = {"model_id": "gpt-4o", "trace_id": "abc123"}
+        mock_dispatch.return_value = _make_response(content="A good response with enough content." * 5)
+
+        # No score_when, no success_when
+        router = Router(goal="test", auto_register=False)
+        router.completion(messages=[{"role": "user", "content": "hi"}])
+
+        # report() should have been called with a heuristic score
+        mock_report.assert_called_once()
+        call_kwargs = mock_report.call_args
+        assert "score" in call_kwargs.kwargs or (len(call_kwargs.args) > 1 if call_kwargs.args else False)
+
+    @patch("kalibr.router.Router._dispatch")
+    @patch("kalibr.router.Router.report")
+    @patch("kalibr.intelligence.decide")
+    def test_success_when_takes_priority_over_default(self, mock_decide, mock_report, mock_dispatch):
+        mock_decide.return_value = {"model_id": "gpt-4o", "trace_id": "abc123"}
+        mock_dispatch.return_value = _make_response(content="no-email-here")
+
+        # success_when provided - should use it, not default scoring
+        router = Router(
+            goal="test",
+            success_when=lambda out: "@" in out,
+            auto_register=False,
+        )
+        router.completion(messages=[{"role": "user", "content": "hi"}])
+
+        # success_when should report success=False (no @ in output), no score kwarg
+        mock_report.assert_called_once_with(success=False)