diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index 12ddfb39a8..a054d7ac53 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -13,7 +13,9 @@ use async_trait::async_trait; use parking_lot::Mutex; use super::fall_through::FallThrough; -use super::llm_class::{LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig}; +use super::llm_class::{ + CustomClassifierConfig, LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig, +}; use super::stage::{StageRouterConfig, build_stage_route}; use super::util::affinity::{ClassifyTrigger, evict_if_full, has_new_user_turn, retention_key}; use super::util::stage::{Tier, set_fall_open}; @@ -35,6 +37,7 @@ struct TierSetter { judge: Arc>, trigger: ClassifyTrigger, message_hash_fallback: bool, + default_tier: Tier, tiers: Mutex>, } @@ -91,6 +94,12 @@ impl Processor for TierSetter { } return Ok(()); } + if self.trigger == ClassifyTrigger::NewSession + && let Some(identity) = identity.as_ref() + { + // A failed opening judgment is still the session's one judgment. + self.retain(identity.clone(), self.default_tier); + } } // Either not this request's turn or the judge had no verdict, so the last // tier stands rather than dropping back to the picker default. @@ -103,10 +112,18 @@ impl Processor for TierSetter { } } +/// Classifier settings for the composite router's retained tier. +pub enum CompositeJudgeConfig { + /// Selects a tier from a capability verdict. + Capability(TaskClassifierConfig), + /// Selects a tier using a custom schema and policy, retaining the fallback on failure. + Custom(CustomClassifierConfig), +} + /// A judge stacked over a stage router. pub struct CompositeRouterConfig { /// Judge settings, including how often `classify_trigger` runs it. - pub judge: TaskClassifierConfig, + pub judge: CompositeJudgeConfig, /// Serves the turns, with the tier the judge picked as its fall-open default. pub stage: StageRouterConfig, } @@ -124,28 +141,39 @@ impl CompositeRouter { /// A stage router carrying its own judge is allowed, but that judge sits ahead /// of the fall-open tier and so answers most of the turns this one set a tier for. pub fn new(config: CompositeRouterConfig) -> Result { - if config.judge.classify_trigger == ClassifyTrigger::EveryRequest { + let (trigger, message_hash_fallback) = match &config.judge { + CompositeJudgeConfig::Capability(judge) => { + (judge.classify_trigger, judge.message_hash_fallback) + } + CompositeJudgeConfig::Custom(config) => { + (config.classify_trigger, config.message_hash_fallback) + } + }; + if trigger == ClassifyTrigger::EveryRequest { return Err(LibsyError::AlgorithmError { message: "composite: classify_trigger must be user_turn or new_session".to_string(), }); } - let trigger = config.judge.classify_trigger; - let message_hash_fallback = config.judge.message_hash_fallback; - // Only the judge's Classifier face is used, so its own affinity never runs. - // Leaving these set would apply the standalone route's pairing rules to a - // trigger this router implements itself. - let judge_config = TaskClassifierConfig { - classify_trigger: ClassifyTrigger::EveryRequest, - message_hash_fallback: false, - ..config.judge + let judge: Arc> = match config.judge { + CompositeJudgeConfig::Capability(judge) => { + // Composite owns the trigger and retention; standalone affinity never runs. + Arc::new(LlmTaskClassifier::new(LlmClassifierConfig::Capability { + config: TaskClassifierConfig { + classify_trigger: ClassifyTrigger::EveryRequest, + message_hash_fallback: false, + ..judge + }, + })?) + } + CompositeJudgeConfig::Custom(config) => { + LlmTaskClassifier::custom_classifier(&config, true)? + } }; - let judge = LlmTaskClassifier::new(LlmClassifierConfig::Capability { - config: judge_config, - })?; let setter = TierSetter { - judge: Arc::new(judge), + judge, trigger, message_hash_fallback, + default_tier: config.stage.mode.default_tier(), tiers: Mutex::new(HashMap::new()), }; let route = build_stage_route(config.stage)? @@ -176,7 +204,7 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; - use switchyard_protocol::{Category, Message, ModelId, Role}; + use switchyard_protocol::{Category, LlmClientError, Message, ModelId, Role}; use super::*; use crate::algorithms::util::stage::PickerMode; @@ -215,7 +243,7 @@ mod tests { fn hash_keyed_router() -> Result> { Ok(Arc::new(CompositeRouter::new(CompositeRouterConfig { - judge: TaskClassifierConfig { + judge: CompositeJudgeConfig::Capability(TaskClassifierConfig { judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { base_threshold: 0.5, ..LlmCapabilityConfig::default() @@ -223,35 +251,160 @@ mod tests { classify_trigger: ClassifyTrigger::UserTurn, message_hash_fallback: true, ..Default::default() - }, + }), stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), })?)) } fn router() -> Result> { Ok(Arc::new(CompositeRouter::new(CompositeRouterConfig { - judge: TaskClassifierConfig { + judge: CompositeJudgeConfig::Capability(TaskClassifierConfig { judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { base_threshold: 0.5, ..LlmCapabilityConfig::default() }), classify_trigger: ClassifyTrigger::UserTurn, ..Default::default() - }, + }), stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), })?)) } #[test] fn rejects_every_request_as_a_trigger() { - let config = CompositeRouterConfig { - judge: TaskClassifierConfig::default(), + for judge in [ + CompositeJudgeConfig::Capability(TaskClassifierConfig::default()), + CompositeJudgeConfig::Custom(CustomClassifierConfig::new( + "Choose a tier", + serde_json::json!({"type": "object"}), + crate::CustomClassifierPolicy::target_selector("/decision/target"), + )), + ] { + let error = CompositeRouter::new(CompositeRouterConfig { + judge, + stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), + }) + .err() + .expect("every_request must be rejected for both classifier modes"); + assert!(matches!(error, LibsyError::AlgorithmError { message } + if message == "composite: classify_trigger must be user_turn or new_session")); + } + } + + fn custom_judge(trigger: ClassifyTrigger) -> CompositeJudgeConfig { + let mut config = CustomClassifierConfig::new( + "Select a tier", + serde_json::json!({"type": "object"}), + crate::CustomClassifierPolicy::target_selector("/decision/target"), + ); + config.classify_trigger = trigger; + CompositeJudgeConfig::Custom(config) + } + + #[tokio::test] + async fn a_failed_session_judgment_retains_the_configured_stage_default() -> Result<()> { + for custom in [false, true] { + for (mode, expected) in [ + (PickerMode::EfficientFirst, "weak"), + (PickerMode::CapableFirst, "strong"), + ] { + let judge = if custom { + custom_judge(ClassifyTrigger::NewSession) + } else { + CompositeJudgeConfig::Capability(TaskClassifierConfig { + classify_trigger: ClassifyTrigger::NewSession, + ..Default::default() + }) + }; + let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { + judge, + stage: StageRouterConfig::new(mode, 0.5), + })?); + let calls = Arc::new(Mutex::new(0)); + let serve = { + let calls = calls.clone(); + move |target: ModelId, _request: Request| { + let completion = if target.as_str() == JUDGE { + *calls.lock() += 1; + "not JSON".to_string() + } else { + target.to_string() + }; + async move { Ok(crate::core::testing::reply(completion)) } + } + }; + for request in [ + user_turn_request(), + turn_request(false), + user_turn_request(), + ] { + let (selected, _) = test_drive_with_models( + router.clone(), + request, + runtime_models(), + serve.clone(), + ) + .await?; + assert_eq!(selected.as_str(), expected); + } + assert_eq!( + *calls.lock(), + 1, + "new_session must retain a failed judgment" + ); + } + } + Ok(()) + } + + #[tokio::test] + async fn a_custom_judge_sets_the_tier_and_retains_it_on_failure() -> Result<()> { + let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { + judge: custom_judge(ClassifyTrigger::UserTurn), stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), - }; - assert!(matches!( - CompositeRouter::new(config), - Err(LibsyError::AlgorithmError { .. }) + })?); + let replies = Arc::new(Mutex::new( + [ + Some(r#"{"decision":{"target":"capable"}}"#), + None, + Some("not JSON"), + Some(r#"{"decision":{"target":"efficient"}}"#), + ] + .into_iter(), )); + let serve = { + let replies = replies.clone(); + move |target: ModelId, _request: Request| { + let completion = if target.as_str() == JUDGE { + replies + .lock() + .next() + .expect("unexpected judge call") + .map(str::to_string) + } else { + Some(target.to_string()) + }; + async move { + completion + .map(crate::core::testing::reply) + .ok_or_else(|| LlmClientError::General("judge unavailable".into())) + } + } + }; + for (request, expected) in [ + (user_turn_request(), "strong"), + (turn_request(false), "strong"), + (user_turn_request(), "strong"), + (user_turn_request(), "strong"), + (user_turn_request(), "weak"), + ] { + let (selected, _) = + test_drive_with_models(router.clone(), request, runtime_models(), serve.clone()) + .await?; + assert_eq!(selected.as_str(), expected); + } + assert!(replies.lock().next().is_none()); + Ok(()) } /// A route may point its judge at the same target it serves capable turns on. diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index a956837902..5aa948101e 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -720,40 +720,49 @@ impl LlmTaskClassifier { } fn build_custom(default_target: Category, config: CustomClassifierConfig) -> Result { + Self::from_classifier( + Self::custom_classifier(&config, false)?, + ClassifierRouteConfig { + default_target, + classify_trigger: config.classify_trigger, + message_hash_fallback: config.message_hash_fallback, + }, + ) + } + + /// The judge face alone, for composite's own trigger and retained tier. + pub(crate) fn custom_classifier( + config: &CustomClassifierConfig, + fail_open: bool, + ) -> Result>> { config.validate()?; - let CustomClassifierConfig { - prompt, - response_schema, - policy, - classify_trigger, - message_hash_fallback, - recent_turn_window, - max_output_tokens, - } = config; - let contract = ClassifierContract::from_inner_schema(&prompt, response_schema)?; - let policy = match policy { + let contract = + ClassifierContract::from_inner_schema(&config.prompt, config.response_schema.clone())?; + let policy = match &config.policy { CustomClassifierPolicy::TargetSelector { selector } => { CustomPolicyRuntime::TargetSelector(TargetSelectorPolicy::new(selector)?) } }; - let classifier: Arc> = Arc::new(JudgeClassifier::new( + let classifier = JudgeClassifier::new( StructuredJudge::new( - TaskInput { recent_turn_window }, + TaskInput { + recent_turn_window: config.recent_turn_window, + }, contract, JsonSchemaDecoder::new(), - JudgeRuntimeConfig::new(max_output_tokens)?, + JudgeRuntimeConfig::new(config.max_output_tokens)?, ), policy, - )); - - Self::from_classifier( - classifier, - ClassifierRouteConfig { - default_target, - classify_trigger, - message_hash_fallback, - }, ) + .with_error_recovery(fail_open); + let classifier = if fail_open { + classifier.with_evidence(|_, verdict| { + verdict.map(|_| serde_json::json!({"source": "llm-classifier"})) + }) + } else { + classifier + }; + Ok(Arc::new(classifier)) } fn build_escalation( diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index efd61f63a9..0c091db879 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -19,7 +19,7 @@ pub use error::{DriverError, LibsyError, Result}; mod algorithms; pub use algorithms::advisor_gate::{AdvisorGate, AdvisorGateConfig, GateTrigger}; -pub use algorithms::composite::{CompositeRouter, CompositeRouterConfig}; +pub use algorithms::composite::{CompositeJudgeConfig, CompositeRouter, CompositeRouterConfig}; pub use algorithms::llm_class::{ CapabilityJudgeConfig, CustomClassifierConfig, CustomClassifierPolicy, DecisionJudgeConfig, LlmCapabilityConfig, LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig, diff --git a/crates/switchyard-runner/src/algorithm.rs b/crates/switchyard-runner/src/algorithm.rs index a994ffc9c6..edf4cc287f 100644 --- a/crates/switchyard-runner/src/algorithm.rs +++ b/crates/switchyard-runner/src/algorithm.rs @@ -11,12 +11,13 @@ use std::sync::Arc; use libsy::{ AdvisorGate, AdvisorGateConfig, Algorithm, BreakerConfig, CapabilityJudgeConfig, - ClassifierContractConfig, ClassifierResponseFormat, ClassifyTrigger, CompositeRouter, - CompositeRouterConfig, CustomClassifierConfig, CustomClassifierPolicy, DecisionJudgeConfig, - EscalationJudgeConfig, GateTrigger, HandoffNoteConfig, LlmCapabilityConfig, - LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, Passthrough, PickerMode, - PlanExecute, PlanExecuteConfig, Random, RuntimeModels, StageRouter, StageRouterConfig, - SubagentRouter, SubagentRouterConfig, TaskClassifierConfig, ToolSemantics, Vgr, VgrConfig, + ClassifierContractConfig, ClassifierResponseFormat, ClassifyTrigger, CompositeJudgeConfig, + CompositeRouter, CompositeRouterConfig, CustomClassifierConfig, CustomClassifierPolicy, + DecisionJudgeConfig, EscalationJudgeConfig, GateTrigger, HandoffNoteConfig, + LlmCapabilityConfig, LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, Passthrough, + PickerMode, PlanExecute, PlanExecuteConfig, Random, RuntimeModels, StageRouter, + StageRouterConfig, SubagentRouter, SubagentRouterConfig, TaskClassifierConfig, ToolSemantics, + Vgr, VgrConfig, }; use serde::Deserialize; use serde_json::Value; @@ -574,6 +575,13 @@ pub enum AdvisorTriggerConfig { pub struct StageClassifierConfig { /// Target the judge is called through. Not a routing destination. pub target: String, + /// Composite accepts custom-schema judges as well as capability judges. + #[serde(default)] + pub mode: Option, + #[serde(default)] + pub response_schema: Option, + #[serde(default)] + pub policy: Option, #[serde(default)] pub decision: Option, #[serde(default)] @@ -645,6 +653,16 @@ impl StageClassifierConfig { tiers: &StageTierConfig, targets: &BTreeMap, ) -> AlgorithmResult { + if self + .mode + .is_some_and(|mode| !matches!(mode, ClassifierMode::Capability)) + || self.response_schema.is_some() + || self.policy.is_some() + { + return Err(AlgorithmConfigError::new( + "custom classifier settings require composite with mode = custom", + )); + } let config = LlmClassifierRouteConfig { classifier_target: self.target.clone(), strong_target: Some(tiers.capable_target.clone()), @@ -668,6 +686,44 @@ impl StageClassifierConfig { _ => Err(AlgorithmConfigError::new("expected a capability judge")), } } + + fn custom_classifier_config( + &self, + route_name: &str, + ) -> AlgorithmResult { + if self.fail_open == Some(false) { + return Err(AlgorithmConfigError::new( + "composite custom classifiers always fail open; remove fail_open = false", + )); + } + if self.decision.is_some() + || self.base_threshold.is_some() + || self.threshold_step.is_some() + || self.response_format_type != ClassifierResponseFormat::JsonSchema + { + return Err(AlgorithmConfigError::new( + "composite custom classifier cannot use capability thresholds, decision settings, or response_format_type", + )); + } + let schema = + required_classifier_field(route_name, "response_schema", &self.response_schema)?; + let schema = serde_json::from_str(&schema).map_err(|error| { + AlgorithmConfigError::with_source( + format!("composite route {route_name}: response_schema is invalid JSON"), + error, + ) + })?; + let mut config = CustomClassifierConfig::new( + required_classifier_field(route_name, "prompt", &self.prompt)?, + schema, + required_classifier_field(route_name, "policy", &self.policy)?.into_libsy(), + ); + config.classify_trigger = self.classify_trigger; + config.message_hash_fallback = self.message_hash_fallback; + config.recent_turn_window = self.recent_turn_window; + config.max_output_tokens = self.max_output_tokens; + Ok(config) + } } impl AlgorithmSpec { @@ -1600,11 +1656,18 @@ fn build_algorithm( } stage_config.tool_semantics = stage.tool_semantics.clone(); stage_config.handoff_notes = stage.handoff_notes.clone(); - let config = CompositeRouterConfig { - judge: classifier.task_classifier_config(route_name, stage, targets)?, - stage: stage_config, + let judge = if matches!(classifier.mode, Some(ClassifierMode::Custom)) { + CompositeJudgeConfig::Custom(classifier.custom_classifier_config(route_name)?) + } else { + CompositeJudgeConfig::Capability( + classifier.task_classifier_config(route_name, stage, targets)?, + ) }; - let algorithm = CompositeRouter::new(config).map_err(|error| { + let algorithm = CompositeRouter::new(CompositeRouterConfig { + judge, + stage: stage_config, + }) + .map_err(|error| { AlgorithmConfigError::with_source( format!("composite route {route_name}: {error}"), error, diff --git a/crates/switchyard-runner/src/config.rs b/crates/switchyard-runner/src/config.rs index b13b1cfa8b..288e4586fc 100644 --- a/crates/switchyard-runner/src/config.rs +++ b/crates/switchyard-runner/src/config.rs @@ -1322,6 +1322,39 @@ evidence = { description = "candidate comparison" } } } + #[test] + fn composite_custom_judge_builds_and_rejects_capability_settings() -> RunnerResult<()> { + let source = composite_config().replace( + "base_threshold = 0.5\nclassify_trigger", + r#"mode = "custom" +prompt = 'Select efficient or capable.' +response_schema = '{"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false}' +policy = { type = "target_selector", selector = "/decision/target" } +classify_trigger"#, + ); + runner_from_toml(&source)?; + runner_from_toml( + &source.replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = true"), + )?; + for (setting, expected) in [ + ( + "base_threshold = 0.5", + "composite custom classifier cannot use capability thresholds", + ), + ( + "fail_open = false", + "composite custom classifiers always fail open", + ), + ] { + let invalid = source.replace( + "mode = \"custom\"", + &format!("mode = \"custom\"\n{setting}"), + ); + assert!(error_message(&invalid).contains(expected)); + } + Ok(()) + } + #[test] fn composite_route_builds_and_claims_both_tiers_and_its_judge() -> RunnerResult<()> { let runner = runner_from_toml(&composite_config())?; diff --git a/docs/routing_algorithms/composite_routing.md b/docs/routing_algorithms/composite_routing.md index 210736f8ec..c13e7e47fc 100644 --- a/docs/routing_algorithms/composite_routing.md +++ b/docs/routing_algorithms/composite_routing.md @@ -59,6 +59,31 @@ nor a second `classifier`. The outer classifier supplies the default tier. Custom `tool_semantics` remain stage-local and have the same additive behavior as on a standalone stage route. +## Custom LLM classifier + +To select a tier directly with your own prompt, keep the stage block and use: + +```toml +[routes.switchyard.classifier] +mode = "custom" +target = "judge" +classify_trigger = "user_turn" +max_output_tokens = 64 +prompt = 'Choose efficient unless capable offers a concrete advantage for this task. Return JSON with decision.target set to efficient or capable.' +response_schema = ''' +{"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false} +''' +policy = { type = "target_selector", selector = "/decision/target" } +``` + +The schema is validated locally. The selector names a tier, and the existing +stage targets supply its model. Custom classifiers always fall open: a missing +or invalid verdict or a client failure retains the previous tier, or uses the +configured stage default for a new session. A failed `new_session` judgment retains +that fallback so tool continuations do not call the judge again. +`fail_open = false` is rejected for custom classifiers. Capability thresholds and +decision-model settings cannot be combined with `mode = "custom"`. + ## Decision model classifier Use a decision target as the classifier. Keep the stage block above