From 0d9e0cbc8e6293a0905ab83ac60779a3431372b1 Mon Sep 17 00:00:00 2001 From: rbehal Date: Tue, 11 Aug 2026 22:32:46 +0000 Subject: [PATCH] feat(python): expose escalation mode to Python escalation(judge_target, efficient_target, capable_target, *, prompt, confirmations, recent_turn_window, window_message_chars, max_output_tokens): each efficient answer is judged and a confirmed streak of escalate verdicts latches the session to the capable target. Sessions ride request metadata (session-id header); a latched session serves capable directly without further judge calls. Version 0.2.0+gumloop.0.2.0. Co-authored-by: Cursor Signed-off-by: rbehal Co-authored-by: Cursor --- crates/switchyard-py/src/libsy_bindings.rs | 67 +++++++++- pyproject.toml | 2 +- switchyard_rust/libsy.py | 13 ++ tests/test_libsy_escalation.py | 146 +++++++++++++++++++++ 4 files changed, 223 insertions(+), 5 deletions(-) create mode 100644 tests/test_libsy_escalation.py diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 0bed73a7..7617700b 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -11,9 +11,9 @@ use pyo3::prelude::*; use serde_json::{Value, json}; use switchyard_libsy::{ Algorithm, ClassifierContractConfig, CustomClassifierConfig, CustomClassifierPolicy, - HandoffNoteConfig, LibsyError as RustLibsyError, LlmClassifierConfig, LlmFallback, LlmTarget, - LlmTargetSet, LlmTaskClassifier, Noop, PickerMode, Random, StageRouter, StageRouterConfig, - TaskClassifierConfig, + EscalationJudgeConfig, HandoffNoteConfig, LibsyError as RustLibsyError, LlmClassifierConfig, + LlmFallback, LlmTarget, LlmTargetSet, LlmTaskClassifier, Noop, PickerMode, Random, StageRouter, + StageRouterConfig, TaskClassifierConfig, }; use switchyard_protocol::{ AggLlmResponse, Context, Decision, LlmClientError, LlmResponse, Metadata, Request, Response, @@ -411,6 +411,61 @@ fn custom_classifier_algorithm( Ok(PyAlgorithm::new(Arc::new(algorithm))) } +/// Construct escalation routing: each efficient answer is judged, and a confirmed streak +/// of escalate verdicts moves the session to the capable target and latches it there. +/// +/// Sessions are identified by request metadata (e.g. a `session-id` header), which +/// `confirmations >= 2` requires since the streak is retained per session. +#[pyfunction(name = "escalation")] +#[pyo3(signature = ( + judge_target, + efficient_target, + capable_target, + *, + prompt=None, + confirmations=None, + recent_turn_window=None, + window_message_chars=None, + max_output_tokens=4096 +))] +#[allow(clippy::too_many_arguments)] +fn escalation_algorithm( + py: Python<'_>, + judge_target: Py, + efficient_target: Py, + capable_target: Py, + prompt: Option, + confirmations: Option, + recent_turn_window: Option, + window_message_chars: Option, + max_output_tokens: u64, +) -> PyResult { + let mut contract = ClassifierContractConfig::default(); + if let Some(prompt) = prompt { + contract = contract.with_prompt(prompt); + } + let mut config = EscalationJudgeConfig::default(); + if let Some(value) = confirmations { + config.confirmations = value; + } + if let Some(value) = recent_turn_window { + config.recent_turn_window = value; + } + if let Some(value) = window_message_chars { + config.window_message_chars = value; + } + let algorithm = LlmTaskClassifier::new(LlmClassifierConfig::Escalation { + judge_target: judge_target.bind(py).try_borrow()?.clone_core(py), + efficient_target: efficient_target.bind(py).try_borrow()?.clone_core(py), + capable_target: capable_target.bind(py).try_borrow()?.clone_core(py), + contract, + config, + max_output_tokens, + }) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + Ok(PyAlgorithm::new(Arc::new(algorithm))) +} + /// Construct signal-driven stage routing with an optional LLM classifier fallback. #[pyfunction(name = "stage_router")] #[pyo3(signature = ( @@ -512,7 +567,11 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { llm_task_classifier_algorithm, &libsy_module )?)?; - libsy_module.add_function(wrap_pyfunction!(custom_classifier_algorithm, &libsy_module)?)?; + libsy_module.add_function(wrap_pyfunction!( + custom_classifier_algorithm, + &libsy_module + )?)?; + libsy_module.add_function(wrap_pyfunction!(escalation_algorithm, &libsy_module)?)?; libsy_module.add_function(wrap_pyfunction!(stage_router_algorithm, &libsy_module)?)?; libsy_module.add("LibsyError", module.getattr("LibsyError")?)?; module.add_submodule(&libsy_module)?; diff --git a/pyproject.toml b/pyproject.toml index 36230371..94550949 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "maturin" [project] # Gumloop fork: dist renamed (mirrors gumloop-celery); import names are unchanged. name = "gumloop-nemo-switchyard" -version = "0.2.0+gumloop.0.1.1" +version = "0.2.0+gumloop.0.2.0" description = "Typed, composable LLM routing with request/response translation and multi-backend orchestration" readme = "README.md" license = "Apache-2.0" diff --git a/switchyard_rust/libsy.py b/switchyard_rust/libsy.py index baa66bca..10bb312e 100644 --- a/switchyard_rust/libsy.py +++ b/switchyard_rust/libsy.py @@ -19,6 +19,7 @@ "LlmTarget", "TaskClassifierConfig", "custom_classifier", + "escalation", "llm_task_classifier", "noop", "random", @@ -118,6 +119,18 @@ def custom_classifier( config: CustomClassifierConfig, ) -> Algorithm: ... + def escalation( + judge_target: LlmTarget, + efficient_target: LlmTarget, + capable_target: LlmTarget, + *, + prompt: str | None = None, + confirmations: int | None = None, + recent_turn_window: int | None = None, + window_message_chars: int | None = None, + max_output_tokens: int = 4096, + ) -> Algorithm: ... + def stage_router( capable_target: LlmTarget, efficient_target: LlmTarget, diff --git a/tests/test_libsy_escalation.py b/tests/test_libsy_escalation.py new file mode 100644 index 00000000..45464a51 --- /dev/null +++ b/tests/test_libsy_escalation.py @@ -0,0 +1,146 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the escalation-mode Python binding. + +Imports go through ``switchyard_rust.libsy`` — the bindings-only surface that must +stay importable without the ``lib`` extra's provider SDKs. +""" + +import json +from typing import Any + +import pytest + +from switchyard_rust.libsy import LlmTarget, escalation + + +def request_body(text: str = "keep going") -> dict[str, Any]: + return { + "model": "auto", + "messages": [ + {"role": "user", "content": [{"type": "text", "text": text}]} + ], + } + + +class EchoClient: + def __init__(self, model: str) -> None: + self.model = model + self.calls: list[dict[str, Any]] = [] + + async def call(self, request: dict[str, Any]) -> dict[str, Any]: + self.calls.append(request) + return { + "model": self.model, + "outputs": [ + { + "role": "assistant", + "content": [{"type": "text", "text": self.model}], + "stop_reason": "end_turn", + } + ], + } + + +class VerdictJudge(EchoClient): + def __init__(self, verdicts: list[bool]) -> None: + super().__init__("judge") + self.verdicts = verdicts + + async def call(self, request: dict[str, Any]) -> dict[str, Any]: + self.calls.append(request) + escalate = self.verdicts.pop(0) + text = json.dumps({"escalate": escalate, "reason": "test verdict"}) + return { + "model": self.model, + "outputs": [ + { + "role": "assistant", + "content": [{"type": "text", "text": text}], + "stop_reason": "end_turn", + } + ], + } + + +def build(judge, efficient, capable, confirmations=1): + return escalation( + LlmTarget("judge", judge), + LlmTarget("efficient", efficient), + LlmTarget("capable", capable), + confirmations=confirmations, + ) + + +async def test_decline_returns_the_efficient_answer_without_a_second_call() -> None: + efficient, capable = EchoClient("efficient"), EchoClient("capable") + algorithm = build(VerdictJudge([False]), efficient, capable) + + _, response = await algorithm.run(request_body()) + + assert response["model"] == "efficient" + assert len(efficient.calls) == 1 + assert capable.calls == [] + + +async def test_confirmed_escalation_moves_to_the_capable_target() -> None: + efficient, capable = EchoClient("efficient"), EchoClient("capable") + algorithm = build(VerdictJudge([True]), efficient, capable, confirmations=1) + + decisions, response = await algorithm.run(request_body()) + + assert response["model"] == "capable" + assert len(efficient.calls) == 1 # judged answer was produced first + assert len(capable.calls) == 1 + assert decisions[-1]["selected_model"] == "capable" + + +async def test_streak_below_confirmations_stays_efficient_then_latches() -> None: + """With confirmations=2 the first escalate verdict stays efficient; the second, + in the same session, latches capable.""" + efficient, capable = EchoClient("efficient"), EchoClient("capable") + judge = VerdictJudge([True, True]) + algorithm = build(judge, efficient, capable, confirmations=2) + headers = {"session-id": "run-42"} + + _, first = await algorithm.run(request_body("step one"), headers=headers) + _, second = await algorithm.run(request_body("step two"), headers=headers) + + assert first["model"] == "efficient" + assert second["model"] == "capable" + # A latched session stops consulting the judge entirely. + _, third = await algorithm.run(request_body("step three"), headers=headers) + assert third["model"] == "capable" + assert len(judge.calls) == 2 + + +async def test_streak_does_not_cross_sessions() -> None: + efficient, capable = EchoClient("efficient"), EchoClient("capable") + judge = VerdictJudge([True, True]) + algorithm = build(judge, efficient, capable, confirmations=2) + + _, first = await algorithm.run(request_body(), headers={"session-id": "run-a"}) + _, second = await algorithm.run(request_body(), headers={"session-id": "run-b"}) + + assert first["model"] == "efficient" + assert second["model"] == "efficient" + + +async def test_judge_failure_stays_on_the_efficient_answer() -> None: + class FailingJudge: + async def call(self, request: dict[str, Any]) -> dict[str, Any]: + raise RuntimeError("judge unavailable") + + efficient, capable = EchoClient("efficient"), EchoClient("capable") + algorithm = build(FailingJudge(), efficient, capable) + + _, response = await algorithm.run(request_body()) + + assert response["model"] == "efficient" + assert capable.calls == [] + + +def test_zero_confirmations_is_rejected() -> None: + with pytest.raises(ValueError, match="confirmations"): + build(EchoClient("judge"), EchoClient("efficient"), EchoClient("capable"), confirmations=0)