diff --git a/crates/libsy-llm-client/README.md b/crates/libsy-llm-client/README.md index d86dc7708..b62da05e1 100644 --- a/crates/libsy-llm-client/README.md +++ b/crates/libsy-llm-client/README.md @@ -12,8 +12,9 @@ headers, makes the call with a shared `reqwest::Client`, and decodes the reply back into a [`switchyard_protocol::Response`] — buffered or streamed. It also pairs the client with a libsy algorithm: [`run`] drives -[`Algorithm::run_stream`] and serves every model call the algorithm offloads, so a -host that just wants the answer never has to drive the step stream itself. +[`Algorithm::run_stream`], serves routing-time calls, and consumes the terminal routing outcome. +When routing has not already produced the answer, `run` makes the terminal call and owns backend +retries plus ordered candidate fallback. It depends on `switchyard-libsy`, `switchyard-protocol`, and `switchyard-translation`; no server, no provider SDK. @@ -143,10 +144,10 @@ async fn stream( ### Routing an algorithm -[`run`] takes a libsy algorithm and a [`ClientRouter`], and returns the final response plus -the trace of decisions the algorithm published. Each offloaded `CallModel` carries an ordered -`models` list. The router resolves and tries those candidates in order; `ClientRouter::single` -is the single-provider case: +[`run`] takes a libsy algorithm and a [`ClientRouter`], and returns the algorithm-selected +[`ModelId`] plus the final response. Routing-time `CallModel`s are served while the algorithm +runs. The terminal outcome either already contains the answer or supplies the selected model and +ordered fallbacks for the client to try. `ClientRouter::single` is the single-provider case: ```rust use std::sync::Arc; @@ -160,12 +161,9 @@ async fn route( request: Request, ) -> switchyard_libsy::Result { let clients = ClientRouter::single(client); - let (trace, _response) = + let (selected_model, _response) = switchyard_llm_client::run(algorithm, clients, request, None).await?; - Ok(trace - .last() - .map(|decision| decision.selected_model_id().to_string()) - .unwrap_or_default()) + Ok(selected_model.to_string()) } ``` diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index e10a17dd6..059675bbe 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -12,9 +12,9 @@ //! streamed responses. //! //! [`run()`] pairs the client with a libsy algorithm: it drives -//! [`switchyard_libsy::Algorithm::run_stream`] and serves every model call the algorithm -//! offloads, so a host that just wants the answer does not have to drive the step stream -//! itself. +//! [`switchyard_libsy::Algorithm::run_stream`], serves routing-time calls, and makes the terminal +//! answer call from the routing outcome when needed. A host that just wants the answer does not +//! have to drive the step stream itself. pub mod backend; pub mod client; @@ -32,3 +32,8 @@ pub use observation::{LlmCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; pub use run::{ClientRouter, run}; pub use switchyard_translation::RawEventStream; + +/// Registers process-wide compatibility gauges with the global meter provider. +pub fn initialize_metrics() { + metrics::initialize(); +} diff --git a/crates/libsy-llm-client/src/metrics.rs b/crates/libsy-llm-client/src/metrics.rs index 582bb5f6b..724e34086 100644 --- a/crates/libsy-llm-client/src/metrics.rs +++ b/crates/libsy-llm-client/src/metrics.rs @@ -4,8 +4,39 @@ //! Metric labelling inherited from Python use std::time::Duration; +use std::{ + sync::OnceLock, + sync::atomic::{AtomicU64, Ordering}, +}; +use opentelemetry::metrics::ObservableGauge; use opentelemetry::{KeyValue, global}; +use switchyard_libsy::Result; +use switchyard_protocol::{ModelId, Response}; + +static TOTAL_REQUESTS: AtomicU64 = AtomicU64::new(0); +static TOTAL_ERRORS: AtomicU64 = AtomicU64::new(0); +static TOTAL_GAUGES: OnceLock<(ObservableGauge, ObservableGauge)> = OnceLock::new(); + +/// Registers process-wide compatibility gauges with the installed global meter provider. +pub fn initialize() { + TOTAL_GAUGES.get_or_init(|| { + let meter = global::meter("switchyard"); + let requests = meter + .u64_observable_gauge("switchyard.total_requests") + .with_callback(|observer| { + observer.observe(TOTAL_REQUESTS.load(Ordering::Relaxed), &[]); + }) + .build(); + let errors = meter + .u64_observable_gauge("switchyard.total_errors") + .with_callback(|observer| { + observer.observe(TOTAL_ERRORS.load(Ordering::Relaxed), &[]); + }) + .build(); + (requests, errors) + }); +} pub(crate) const fn is_retryable_http_status(status: u16) -> bool { status == 408 || status == 429 || (status >= 500 && status <= 599) @@ -61,16 +92,9 @@ pub(crate) fn record_upstream_attempt(status: Option) { ); } -/// Records what routing cost on top of the call that served the run: classifier -/// calls, target resolution, and decision publishing. -pub(crate) fn record_routing_overhead( - algorithm: &str, - run: Duration, - call_duration: Duration, -) -> Duration { - // Saturating: the two clocks start a moment apart, so a run that is all - // routed call can come out fractionally negative. - let overhead = run.saturating_sub(call_duration); +/// Records the time needed to produce the routing outcome, including classifier calls, +/// target resolution, request rewrites, and decision publishing. +pub(crate) fn record_routing_overhead(algorithm: &str, overhead: Duration) { global::meter("switchyard") .f64_histogram("switchyard.routing_overhead_ms") .build() @@ -78,7 +102,58 @@ pub(crate) fn record_routing_overhead( overhead.as_secs_f64() * 1000.0, &[KeyValue::new("algorithm", algorithm.to_string())], ); - overhead +} + +/// Records one terminal model call made after routing, preserving the libsy call metric surface. +pub(crate) fn record_answer_call( + algorithm: &str, + selected_model: &ModelId, + duration: Duration, + result: &Result, +) { + let attributes = [ + KeyValue::new("algorithm", algorithm.to_string()), + KeyValue::new("selected_model", selected_model.to_string()), + KeyValue::new("outcome", if result.is_ok() { "ok" } else { "error" }), + ]; + let meter = global::meter("switchyard"); + meter + .u64_counter("switchyard.llm_calls") + .build() + .add(1, &attributes); + meter + .f64_histogram("switchyard.llm_call_duration_ms") + .build() + .record(duration.as_secs_f64() * 1000.0, &attributes); +} + +/// Records one terminal routed request after the algorithm has produced an outcome. +pub(crate) fn record_routed_request( + selected_model: &ModelId, + answer_duration: Option, + result: &Result, +) { + TOTAL_REQUESTS.fetch_add(1, Ordering::Relaxed); + let attributes = [KeyValue::new("model", selected_model.to_string())]; + let meter = global::meter("switchyard"); + if result.is_ok() { + meter + .u64_counter("switchyard.requests") + .build() + .add(1, &attributes); + if let Some(duration) = answer_duration { + meter + .f64_histogram("switchyard.model_call_latency_ms") + .build() + .record(duration.as_secs_f64() * 1000.0, &attributes); + } + } else { + TOTAL_ERRORS.fetch_add(1, Ordering::Relaxed); + meter + .u64_counter("switchyard.errors") + .build() + .add(1, &attributes); + } } #[cfg(test)] diff --git a/crates/libsy-llm-client/src/observation.rs b/crates/libsy-llm-client/src/observation.rs index cff08fff4..da448eaf2 100644 --- a/crates/libsy-llm-client/src/observation.rs +++ b/crates/libsy-llm-client/src/observation.rs @@ -8,13 +8,11 @@ use std::time::Duration; use switchyard_protocol::{ModelId, Usage}; -/// One completed model call observed at the algorithm offload boundary. +/// One completed model call observed while serving an algorithm run. #[derive(Clone, Debug)] pub struct LlmCallObservation { /// Model selected for the completed call. pub selected_model: ModelId, - /// Whether this call generated an answer rather than a routing verdict. - pub is_answer_call: bool, /// Whether the call completed successfully. pub is_success: bool, /// Time spent waiting for the model call to resolve. @@ -26,8 +24,10 @@ pub struct LlmCallObservation { /// One request-scoped observation emitted by the algorithm runner. #[derive(Clone, Debug)] pub enum RunObservation { - /// A completed model call. + /// A completed model call requested by the algorithm for routing work. LlmCall(LlmCallObservation), + /// A completed terminal model call made from the routing outcome. + AnswerCall(LlmCallObservation), /// Routing time recorded by the `switchyard.routing_overhead_ms` metric. RoutingOverhead(Duration), } diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 0319f1d8a..73d897c39 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -5,8 +5,8 @@ //! //! [`switchyard_libsy::Algorithm::run_stream`] is the whole libsy API: it yields a stream of //! steps and expects its consumer to serve every offloaded model call. [`run()`] is that -//! consumer — it drives the stream with [`switchyard_libsy::drive`], hands each call to a -//! [`RoutedLlmClient`], and returns the final response with the trace of decisions. +//! consumer — it drives the stream with [`switchyard_libsy::drive`], hands routing-time calls to +//! a [`RoutedLlmClient`], and serves the terminal routing outcome. //! //! libsy owns the stream mechanics; what this module adds is ordered candidate fallback and the //! `libsy.client_call` span around each candidate. Each candidate exhausts its backend retry @@ -15,13 +15,13 @@ use std::collections::HashMap; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; use http::StatusCode; use parking_lot::Mutex; use switchyard_libsy::{Algorithm, CallModel, LibsyError, Result, drive}; use switchyard_protocol::{ - Decision, LlmClientError, ModelId, Request, Response, RoutedLlmClient, RoutingFallbackReason, + LlmClientError, ModelId, Request, Response, RoutedLlmClient, RoutingFallbackReason, }; use crate::observation::{LlmCallObservation, RunObservation, RunObserver}; @@ -29,77 +29,99 @@ use crate::{metrics, observability}; /// Run one request to completion, serving every offloaded model call with `client`. /// -/// Returns the final [`Response`] and the trace of [`Decision`]s the algorithm published along -/// the way. `observer`, when present, receives each completed model call and, after a -/// successful routed run, its routing overhead. +/// Returns the model selected by the algorithm and the final [`Response`]. `observer`, when +/// present, receives each completed routing or answer call and the routing overhead. /// /// `clients` resolves each offloaded call to the client for the target the algorithm /// selected — an algorithm may route among targets served by different providers, so this is /// a per-call lookup, not one client for the whole run. Use /// [`ClientRouter::single`](ClientRouter::single) when one client serves every target. /// -/// A failed *model* call is forwarded back into the algorithm, which may route around it; -/// this returns `Err` only when the run itself cannot complete. +/// Routing-time model failures are forwarded back into the algorithm. Once routing completes, +/// this client exhausts backend retries and then the outcome's ordered candidate fallbacks. pub async fn run( algorithm: Arc, clients: ClientRouter, request: Request, observer: Option, -) -> Result<(Vec, Response)> { +) -> Result<(ModelId, Response)> { let algorithm_name = algorithm.name().to_string(); - // The output from `serve` goes in here: when each successful routed call was in - // flight. Everything else the run spent time on is routing overhead. - let routed_calls = Arc::new(Mutex::new(RoutedCallWindows::default())); let run_started = Instant::now(); - let result = drive(algorithm, request, { - let observer = observer.clone(); - let routed_calls = Arc::clone(&routed_calls); - move |call| { - serve( - clients.clone(), - call, - observer.clone(), - Arc::clone(&routed_calls), - ) - } + let routing_clients = clients.clone(); + // This says if we have an observer, put Some(..) in routing_observations. + // No observer means we don't want any routing_observations. + let routing_observations = observer.as_ref().map(|_| Arc::new(Mutex::new(Vec::new()))); + let outcome = drive(algorithm, request, { + let routing_observations = routing_observations.clone(); + move |call| serve(routing_clients.clone(), call, routing_observations.clone()) }) - .await?; - if let Some(served) = routed_calls.lock().served() { - let overhead = - metrics::record_routing_overhead(&algorithm_name, run_started.elapsed(), served); - if let Some(observer) = observer { - observer(RunObservation::RoutingOverhead(overhead)); - } + .await; + let answered_model = outcome + .as_ref() + .ok() + .and_then(|outcome| outcome.response.as_ref()) + .and_then(Response::served_model); + emit_routing_observations(&observer, &routing_observations, answered_model); + let outcome = outcome?; + let overhead = run_started.elapsed(); + metrics::record_routing_overhead(&algorithm_name, overhead); + + let selected_model_id = outcome.selected_model_id; + let (result, answer_duration) = if let Some(response) = outcome.response { + (Ok(response), None) + } else { + let mut models = Vec::with_capacity(1 + outcome.fallback_models.len()); + models.push(selected_model_id.clone()); + models.extend(outcome.fallback_models); + let answer_started = Instant::now(); + let observe = |observation| { + if let Some(observer) = &observer { + observer(RunObservation::AnswerCall(observation)); + } + }; + let result = call_first_available( + &clients, + &algorithm_name, + &outcome.request, + &models, + &observe, + ) + .await; + let answer_duration = answer_started.elapsed(); + metrics::record_answer_call( + &algorithm_name, + &selected_model_id, + answer_duration, + &result, + ); + (result, Some(answer_duration)) + }; + metrics::record_routed_request(&selected_model_id, answer_duration, &result); + if let Some(observer) = &observer { + observer(RunObservation::RoutingOverhead(overhead)); } - Ok(result) + result.map(|response| (selected_model_id, response)) } -/// The wall-clock windows during which a successful routed call was in flight. -/// An algorithm can make multiple overlapping calls. This is the union of all of them. -#[derive(Default)] -struct RoutedCallWindows(Vec<(Instant, Instant)>); - -impl RoutedCallWindows { - /// Record one completed routed call. - fn record(&mut self, started: Instant, ended: Instant) { - self.0.push((started, ended)); - } - - /// Total time at least one routed call was in flight, merging overlapping windows. - fn served(&mut self) -> Option { - self.0.sort_unstable_by_key(|(started, _)| *started); - // Sweep the windows in start order, advancing a cursor along the timeline and - // counting only the time each window covers that the cursor has not reached. - let mut covered = self.0.first()?.0; - let mut total = Duration::ZERO; - for &(started, ended) in &self.0 { - covered = covered.max(started); - if ended > covered { - total += ended - covered; - covered = ended; - } +/// Emits completed routing calls after the outcome reveals whether one response became the answer. +fn emit_routing_observations( + observer: &Option, + observations: &Option>>>, + answered_model: Option<&ModelId>, +) { + let (Some(observer), Some(observations)) = (observer, observations) else { + return; + }; + let mut answer_observation = None; + for observation in observations.lock().drain(..) { + if answer_observation.is_none() && answered_model == Some(&observation.selected_model) { + answer_observation = Some(observation); + } else { + observer(RunObservation::LlmCall(observation)); } - Some(total) + } + if let Some(observation) = answer_observation { + observer(RunObservation::AnswerCall(observation)); } } @@ -110,41 +132,51 @@ impl RoutedCallWindows { async fn serve( clients: ClientRouter, call: CallModel, - observer: Option, - // Output parameter because `drive` takes a function that returns a plain `Result<()>`. - routed_calls: Arc>, + observations: Option>>>, ) -> Result<()> { - let result = call_first_available(&clients, &call, &observer, &routed_calls).await; + let observe = |observation| { + if let Some(observations) = &observations { + observations.lock().push(observation); + } + }; + let result = call_first_available( + &clients, + &call.algorithm, + &call.request, + &call.models, + &observe, + ) + .await; call.respond(result) } /// Try candidates in order until one succeeds or a failure stops fallback. async fn call_first_available( clients: &ClientRouter, - call: &CallModel, - observer: &Option, - routed_calls: &Arc>, + algorithm: &str, + request: &Request, + models: &[ModelId], + observe: &(dyn Fn(LlmCallObservation) + Send + Sync), ) -> Result { - for (index, target) in call.models.iter().enumerate() { - let request = request_for(&call.request, target); + for (index, target) in models.iter().enumerate() { + let request = request_for(request, target); match call_one( clients, target, request, - call, - observer, - routed_calls, + algorithm, + observe, index, - call.models.len(), + models.len(), ) .await { Ok(response) => return Ok(response), - Err(error) if index + 1 == call.models.len() => return Err(error), + Err(error) if index + 1 == models.len() => return Err(error), Err(error) => match fallback_reason(&error) { Some(reason) => tracing::info!( from = %target, - to = %call.models[index + 1], + to = %models[index + 1], reason = reason.as_str(), "model call failed; trying next candidate" ), @@ -162,8 +194,8 @@ async fn call_first_available( name = "libsy.client_call", skip_all, fields( - algorithm = call.algorithm, - switchyard.algorithm = call.algorithm, + algorithm = algorithm, + switchyard.algorithm = algorithm, switchyard.candidate = index + 1, switchyard.candidate_count = count, selected_model = %model_id, @@ -199,9 +231,8 @@ async fn call_one( clients: &ClientRouter, model_id: &ModelId, request: Request, - call: &CallModel, - observer: &Option, - routed_calls: &Arc>, + algorithm: &str, + observe: &(dyn Fn(LlmCallObservation) + Send + Sync), // index is for span log index: usize, // count is for span log @@ -209,15 +240,13 @@ async fn call_one( ) -> Result { let span = tracing::Span::current(); observability::record_gen_ai_request(&span, &request.llm_request); - if let Some(session_id) = call - .request + if let Some(session_id) = request .metadata .as_ref() .and_then(|metadata| metadata.session_id.as_deref()) { span.record("gen_ai.conversation.id", session_id); } - let is_answer_call = call.is_answer_call; // Resolved before the clock starts: picking the client is Switchyard's work, not // the provider's, so it belongs in the routing overhead. let client = clients.route(model_id); @@ -227,31 +256,23 @@ async fn call_one( Err(error) => Err(error), } .map_err(|source| LibsyError::client_call(model_id.clone(), source)); - let ended = Instant::now(); - let duration = ended - started; + let duration = started.elapsed(); let result = result.map(|mut response| { response.set_served_model(model_id); response }); let result = observability::observe_client_call(result); - if let Some(observer) = observer { - observer(RunObservation::LlmCall(LlmCallObservation { - selected_model: model_id.clone(), - is_answer_call, - is_success: result.is_ok(), - duration, - usage: result - .as_ref() - .ok() - .and_then(|response| response.llm_response.as_agg()) - .map(|response| response.usage.clone()), - })); - } - if is_answer_call && result.is_ok() { - routed_calls.lock().record(started, ended); - } - + observe(LlmCallObservation { + selected_model: model_id.clone(), + is_success: result.is_ok(), + duration, + usage: result + .as_ref() + .ok() + .and_then(|response| response.llm_response.as_agg()) + .map(|response| response.usage.clone()), + }); result } @@ -358,7 +379,7 @@ mod tests { use async_trait::async_trait; use futures::StreamExt; use http::StatusCode; - use switchyard_libsy::Driver; + use switchyard_libsy::{Driver, RoutingOutcome}; use switchyard_protocol::{ LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, completion_text, text_request, text_response, @@ -372,14 +393,49 @@ mod tests { models: Vec, } + struct AnsweredAlgorithm { + model: ModelId, + } + #[async_trait] impl Algorithm for CandidateAlgorithm { fn name(&self) -> &str { "candidate_test" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { - driver.call_model(request, self.models.clone(), true).await + async fn route( + self: Arc, + _driver: Driver, + request: Request, + ) -> Result { + let selected_model = self.models.first().cloned().ok_or(LibsyError::NoTargets)?; + Ok(RoutingOutcome::route_to( + selected_model, + self.models.iter().skip(1).cloned().collect(), + request, + )) + } + } + + #[async_trait] + impl Algorithm for AnsweredAlgorithm { + fn name(&self) -> &str { + "answered_test" + } + + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { + let response = driver + .call_model(request.clone(), vec![self.model.clone()]) + .await?; + Ok(RoutingOutcome::answered( + self.model.clone(), + request, + response, + )) } } @@ -462,7 +518,7 @@ mod tests { async fn run_candidates( first: FirstOutcome, - ) -> (Arc, Result<(Vec, Response)>) { + ) -> (Arc, Result<(ModelId, Response)>) { let client = Arc::new(CandidateClient { calls: Mutex::new(Vec::new()), first, @@ -480,6 +536,39 @@ mod tests { (client, result) } + #[tokio::test] + async fn answered_outcome_does_not_make_a_second_model_call() -> Result<()> { + let client = Arc::new(CandidateClient { + calls: Mutex::new(Vec::new()), + first: FirstOutcome::StreamSuccess, + }); + let observations = Arc::new(Mutex::new(Vec::new())); + let observed = Arc::clone(&observations); + let observer: RunObserver = Arc::new(move |event| observed.lock().push(event)); + + let (selected, response) = run( + Arc::new(AnsweredAlgorithm { + model: "weak".into(), + }), + ClientRouter::single(client.clone()), + request(), + Some(observer), + ) + .await?; + + assert_eq!(selected, "weak"); + assert_eq!(response.served_model().map(ModelId::as_str), Some("weak")); + assert_eq!(&*client.calls.lock(), &[ModelId::from("weak")]); + let observations = observations.lock(); + assert!(matches!(observations[0], RunObservation::AnswerCall(_))); + assert!(matches!( + observations[1], + RunObservation::RoutingOverhead(_) + )); + assert_eq!(observations.len(), 2); + Ok(()) + } + #[test] fn fallback_only_accepts_context_and_unavailable_failures() { let error = |source| LibsyError::client_call("target", source); diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index bf37300ef..60f871783 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -12,6 +12,7 @@ use std::collections::BTreeMap; use std::fmt; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, OnceLock}; use std::time::Duration; @@ -34,12 +35,13 @@ use tracing_subscriber::registry::LookupSpan; use switchyard_libsy::{ AffinityRouter, Algorithm, Classifier, Driver, LibsyError, LlmClassifierConfig, - LlmTaskClassifier, PickerMode, StageRouter, StageRouterConfig, Step, TaskClassifierConfig, + LlmTaskClassifier, PickerMode, RoutingOutcome, StageRouter, StageRouterConfig, Step, + TaskClassifierConfig, }; use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::ModelId; use switchyard_protocol::{ - ContentBlock, Decision, LlmRequest, LlmResponse, Message, Metadata, Request, Response, Role, + ContentBlock, LlmRequest, LlmResponse, Message, Metadata, Request, Response, Role, RoutedLlmClient, ToolCall, ToolResult, Usage, WireFormat, }; use switchyard_protocol::{ @@ -182,7 +184,7 @@ fn telemetry() -> &'static Telemetry { let reader = PeriodicReader::builder(exporter.clone()).build(); let provider = SdkMeterProvider::builder().with_reader(reader).build(); opentelemetry::global::set_meter_provider(provider.clone()); - switchyard_libsy::initialize_metrics(); + switchyard_llm_client::initialize_metrics(); let span_exporter = InMemorySpanExporter::default(); let tracer_provider = SdkTracerProvider::builder() @@ -332,6 +334,39 @@ struct ClassifierClient { routed_delay: Duration, } +/// Fails the first efficient answer so candidate fallback can be compared with affinity. +struct AffinityFallbackClient { + calls: Mutex>, + efficient_available: AtomicBool, +} + +#[async_trait] +impl RoutedLlmClient for AffinityFallbackClient { + async fn call(&self, request: Request) -> Result { + let model = request.model_id().unwrap_or_default(); + self.calls.lock().push(model.clone()); + if model == "affinity-fallback-judge" { + return Ok(Response { + llm_response: LlmResponse::Agg(text_response( + Some(model.to_string()), + r#"{"crux":"bounded task","primary_rule":"SUP-1","capability_boundary":"supported","p_solve":0.9}"#, + )), + metadata: None, + }); + } + if model == "affinity-fallback-weak" && !self.efficient_available.load(Ordering::Relaxed) { + return Err(LlmClientError::ContextWindowExceeded { + model, + message: "too long".to_string(), + }); + } + Ok(Response { + llm_response: LlmResponse::Agg(text_response(Some(model.to_string()), "answer")), + metadata: None, + }) + } +} + #[async_trait] impl RoutedLlmClient for ClassifierClient { async fn call(&self, request: Request) -> Result { @@ -418,8 +453,7 @@ impl RoutedLlmClient for UsageClient { } } -/// Publishes one decision for the first target, then calls it — the smallest -/// algorithm exercising both instrumented driver paths. +/// Selects the first target and leaves the terminal call to the client. struct SingleCallAlgo { name: String, target_set: Vec, @@ -433,18 +467,44 @@ impl Algorithm for SingleCallAlgo { async fn route( self: Arc, - driver: Driver, + _driver: Driver, request: Request, - ) -> switchyard_libsy::Result { + ) -> switchyard_libsy::Result { let target = self .target_set .first() .ok_or(LibsyError::NoTargets)? .clone(); tracing::info!("picked '{target}'"); - let decision = Decision::new(target.clone(), true); - driver.decide(decision.clone()).await?; - driver.call_model(request, vec![target.into()], true).await + Ok(RoutingOutcome::route_to(target.into(), Vec::new(), request)) + } +} + +/// Makes one routing-time model call and returns its response as the answer. +struct RoutingCallAlgo { + name: String, + target: ModelId, +} + +#[async_trait] +impl Algorithm for RoutingCallAlgo { + fn name(&self) -> &str { + &self.name + } + + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> switchyard_libsy::Result { + let response = driver + .call_model(request.clone(), vec![self.target.clone()]) + .await?; + Ok(RoutingOutcome::answered( + self.target.clone(), + request, + response, + )) } } @@ -479,7 +539,7 @@ async fn run( algorithm: Arc, client: Arc, request: Request, -) -> switchyard_libsy::Result<(Vec, Response)> { +) -> switchyard_libsy::Result<(ModelId, Response)> { switchyard_llm_client::run(algorithm, ClientRouter::single(client), request, None).await } @@ -602,6 +662,60 @@ async fn affinity_warns_once_when_request_has_no_usable_identity() -> switchyard Ok(()) } +#[tokio::test] +async fn affinity_keeps_the_algorithm_selection_after_client_fallback() +-> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let client = Arc::new(AffinityFallbackClient { + calls: Mutex::new(Vec::new()), + efficient_available: AtomicBool::new(false), + }); + let router = Arc::new(LlmTaskClassifier::new(LlmClassifierConfig::Capability { + judge_target: "affinity-fallback-judge".into(), + efficient_target: "affinity-fallback-weak".into(), + capable_target: "affinity-fallback-strong".into(), + config: TaskClassifierConfig { + base_threshold: 0.5, + session_affinity: true, + ..TaskClassifierConfig::default() + }, + })?) as Arc; + let request = request_with_metadata("affinity-fallback-session", "affinity-fallback-first"); + + let (selected, first_response) = switchyard_llm_client::run( + Arc::clone(&router), + ClientRouter::single(client.clone()), + request.clone(), + None, + ) + .await?; + assert_eq!(selected, "affinity-fallback-weak"); + assert_eq!( + first_response.served_model().map(ModelId::as_str), + Some("affinity-fallback-strong") + ); + + client.efficient_available.store(true, Ordering::Relaxed); + let (selected, second_response) = + switchyard_llm_client::run(router, ClientRouter::single(client.clone()), request, None) + .await?; + assert_eq!(selected, "affinity-fallback-weak"); + assert_eq!( + second_response.served_model().map(ModelId::as_str), + Some("affinity-fallback-weak") + ); + assert_eq!( + &*client.calls.lock(), + &[ + ModelId::from("affinity-fallback-judge"), + ModelId::from("affinity-fallback-weak"), + ModelId::from("affinity-fallback-strong"), + ModelId::from("affinity-fallback-weak"), + ] + ); + Ok(()) +} + #[tokio::test] async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; @@ -630,8 +744,8 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l request.llm_request.output.max_output_tokens = Some(512); request.llm_request.output.response_format = Some(json!({"type": "json_schema"})); request.llm_request.reasoning.effort = Some("high".to_string()); - let (trace, _response) = run(algo(ALGO, MODEL), client, request).await?; - assert_eq!(trace.len(), 1); + let (selected_model, _response) = run(algo(ALGO, MODEL), client, request).await?; + assert_eq!(selected_model, MODEL); // Metrics: run/call counters and latency histograms keyed by algorithm, // plus one published decision. @@ -648,11 +762,11 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l Some(1) ); assert_eq!( - u64_counter_value(&snapshots, "switchyard.llm_calls", &call_attrs), + f64_histogram_count(&snapshots, "switchyard.run_duration_ms", &run_attrs), Some(1) ); assert_eq!( - f64_histogram_count(&snapshots, "switchyard.run_duration_ms", &run_attrs), + u64_counter_value(&snapshots, "switchyard.llm_calls", &call_attrs), Some(1) ); assert_eq!( @@ -803,33 +917,6 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l Some(&OtelValue::I64(18)) ); - let call_span = find_span(&spans, "libsy.llm_call", "selected_model", MODEL); - assert_eq!(call_span.parent.as_deref(), Some("libsy.run")); - assert_eq!( - call_span.fields.get("algorithm").map(String::as_str), - Some(ALGO) - ); - assert_eq!( - call_span.fields.get("outcome").map(String::as_str), - Some("ok") - ); - assert_eq!( - call_span.fields.get("input_tokens").map(String::as_str), - Some("11") - ); - assert_eq!( - call_span.fields.get("output_tokens").map(String::as_str), - Some("7") - ); - assert_eq!( - call_span.fields.get("total_tokens").map(String::as_str), - Some("25") - ); - assert_eq!( - call_span.fields.get("reasoning_tokens").map(String::as_str), - Some("2") - ); - // The algorithm logs why it made the decision. let events = store.events(); assert!( @@ -894,8 +981,8 @@ async fn stage_router_records_algorithm_owned_metrics() -> switchyard_libsy::Res usage: Usage::default(), }) as Arc; - let (trace, _) = run(algorithm, client, request).await?; - assert_eq!(trace[0].selected_model_id(), STRONG); + let (selected_model, _) = run(algorithm, client, request).await?; + assert_eq!(selected_model, STRONG); let snapshots = flushed_metrics(exporter, provider); assert_eq!( @@ -948,11 +1035,10 @@ async fn observed_run_reports_one_successful_routed_call() -> switchyard_libsy:: ); let observations = observations.lock(); assert_eq!(observations.len(), 2); - let RunObservation::LlmCall(observation) = &observations[0] else { - return Err(test_error("expected an LLM call observation")); + let RunObservation::AnswerCall(observation) = &observations[0] else { + return Err(test_error("expected an answer-call observation")); }; assert_eq!(observation.selected_model, MODEL); - assert!(observation.is_answer_call); assert!(observation.is_success); assert!(observation.usage.is_some()); assert!(matches!( @@ -1105,14 +1191,12 @@ async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy:: let (store, exporter, provider, _, _) = telemetry(); const ALGO: &str = "obs-failure-algo"; const MODEL: &str = "obs-failure-model"; - let before = flushed_metrics(exporter, provider); - let total_requests_before = - u64_gauge_value(&before, "switchyard.total_requests").unwrap_or_default(); - let total_errors_before = - u64_gauge_value(&before, "switchyard.total_errors").unwrap_or_default(); - - // The call is offloaded and we fail it by hand, without a client. - let stream = algo(ALGO, MODEL).run_stream(request_with_metadata("obs-session-2", "obs-corr-2")); + // The routing call is offloaded and we fail it by hand, without a client. + let algorithm = Arc::new(RoutingCallAlgo { + name: ALGO.to_string(), + target: MODEL.into(), + }); + let stream = algorithm.run_stream(request_with_metadata("obs-session-2", "obs-corr-2")); tokio::pin!(stream); let mut saw_error_step = false; @@ -1121,7 +1205,6 @@ async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy:: Ok(Step::CallModel(call)) => { call.respond(Err(test_error("synthetic upstream failure")))?; } - Ok(Step::Decision(_)) => {} Ok(Step::Done(_)) => { return Err(test_error("expected the failed call to fail the run")); } @@ -1133,7 +1216,8 @@ async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy:: "expected an error step from the failed call" ); - // Metrics: the call and the run both count under outcome=error. + // The routing call and algorithm run fail, but no terminal outcome exists to attribute as a + // routed request. let snapshots = flushed_metrics(exporter, provider); let run_attrs = [("algorithm", ALGO), ("outcome", "error")]; let call_attrs = [ @@ -1149,18 +1233,6 @@ async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy:: u64_counter_value(&snapshots, "switchyard.llm_calls", &call_attrs), Some(1) ); - assert_eq!( - u64_counter_value(&snapshots, "switchyard.errors", &[("model", MODEL)]), - Some(1) - ); - assert_eq!( - u64_gauge_value(&snapshots, "switchyard.total_requests"), - Some(total_requests_before + 1) - ); - assert_eq!( - u64_gauge_value(&snapshots, "switchyard.total_errors"), - Some(total_errors_before + 1) - ); // Nothing was served, so there is nothing to measure routing against. assert_eq!( f64_histogram_count( @@ -1220,7 +1292,7 @@ async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy:: } #[tokio::test] -async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_libsy::Result<()> { +async fn classifier_metrics_count_routing_and_answer_calls_once() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; let (_store, exporter, provider, _, _) = telemetry(); let before = flushed_metrics(exporter, provider); @@ -1234,14 +1306,9 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib }) as Arc; let router = classifier_router("classifier", "weak", "strong")?; - let (trace, _response) = run(router, client, classifier_request()).await?; + let (selected_model, _response) = run(router, client, classifier_request()).await?; - assert_eq!( - trace - .last() - .map(|decision| decision.selected_model_id().as_str()), - Some("weak") - ); + assert_eq!(selected_model, "weak"); let snapshots = flushed_metrics(exporter, provider); assert_eq!( @@ -1256,6 +1323,18 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib ), Some(1) ); + assert_eq!( + u64_counter_value( + &snapshots, + "switchyard.llm_calls", + &[ + ("algorithm", "llm_task_classifier"), + ("selected_model", "weak"), + ("outcome", "ok"), + ], + ), + Some(1) + ); assert_eq!( u64_counter_value(&snapshots, "switchyard.requests", &[("model", "weak")],), Some(1) diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 08072689c..b69c69754 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -21,7 +21,7 @@ tokio = { version = "1", features = ["macros", "rt"] } | Type | Purpose | |---|---| -| [`Passthrough`] | Always call one configured target. | +| [`Passthrough`] | Always select one configured target. | | [`Random`] | Select among any number of targets using uniform or weighted routing. | | [`LlmTaskClassifier`] | Ask a judge model to choose an efficient or capable target. | | [`StageRouter`] | Route coding-agent turns from tool and progress signals, with an optional judge fallback. | @@ -30,13 +30,13 @@ tokio = { version = "1", features = ["macros", "rt"] } ## How it fits together -A target is a bare model id naming a routing destination. An [`Algorithm`] selects targets and -records [`Decision`](switchyard_protocol::Decision)s, offloading every model call -to its caller: [`Algorithm::run_stream`] yields a [`Step`] stream whose -[`Step::CallModel`] items the host serves over its own transport. Each call carries -an ordered, non-empty list of candidate models; the host tries them until one -answers. libsy makes no network calls itself — `switchyard-llm-client`'s `run` is -a ready-made consumer that drives the stream and performs the calls over HTTP. +A target is a bare model id naming a routing destination. An [`Algorithm`] may offload +routing-time classifier or judge calls to its caller: [`Algorithm::run_stream`] yields a [`Step`] +stream whose [`Step::CallModel`] items the host serves over its own transport. The stream ends +with [`Step::Done`] carrying a [`RoutingOutcome`]: the algorithm-selected model, ordered +fallbacks, rewritten request, and an optional response already produced while routing. libsy +makes no network calls itself — `switchyard-llm-client`'s `run` is a ready-made consumer that +drives the stream and performs the terminal answer call, retries, and fallback over HTTP. The provider-neutral [`Request`], [`Response`], [`Usage`], and [`LlmResponse`] contracts come from `switchyard-protocol`. diff --git a/crates/libsy/src/algorithms/advisor_gate.rs b/crates/libsy/src/algorithms/advisor_gate.rs index 3d3cb409b..63cabdd16 100644 --- a/crates/libsy/src/algorithms/advisor_gate.rs +++ b/crates/libsy/src/algorithms/advisor_gate.rs @@ -32,11 +32,11 @@ use std::time::Instant; use parking_lot::Mutex; use switchyard_protocol::{ - ContentBlock, Decision, InstructionBlock, LlmRequest, Message, ModelId, OutputParams, Request, - Response, Role, SamplingParams, + ContentBlock, InstructionBlock, LlmRequest, Message, ModelId, OutputParams, Request, Role, + SamplingParams, }; -use crate::core::algorithm::{Algorithm, Driver}; +use crate::core::algorithm::{Algorithm, Driver, RoutingOutcome}; use crate::{LibsyError, Result}; mod telemetry; @@ -233,12 +233,6 @@ impl AdvisorGate { }) } - /// One executor Decision; published immediately before each executor call - /// so `trace.last()` always names the executor on every return path. - fn executor_decision(&self) -> Decision { - Decision::new(self.executor.clone(), true) - } - // ── Scope ledger ──────────────────────────────────────────────────────── /// Whether the scope's budget or failure cap is spent; logs once per scope. @@ -324,23 +318,23 @@ impl AdvisorGate { driver: &Driver, request: Request, scope: &ScopeKey, - ) -> Result { + ) -> Result { // Spent budget (or failure cap): pure passthrough — live stream, // verbatim preserved-body replay, zero buffering. Executor errors // (including ContextWindowExceeded) propagate for the host's // client-visible mapping. if self.check_exhausted(scope) { - driver.decide(self.executor_decision()).await?; - return driver - .call_model(request, vec![self.executor.clone()], true) - .await; + return Ok(RoutingOutcome::route_to( + self.executor.clone(), + Vec::new(), + request, + )); } // Gated phase: generate the turn once, fully buffered, so the gate // can inspect it before the client sees anything. - driver.decide(self.executor_decision()).await?; let response = driver - .call_model(request.clone(), vec![self.executor.clone()], true) + .call_model(request.clone(), vec![self.executor.clone()]) .await?; let turn = buffer_turn(self.executor.as_str(), response).await?; @@ -362,7 +356,11 @@ impl AdvisorGate { } }; if !(triggered || stall) { - return Ok(turn.into_response()); + return Ok(RoutingOutcome::answered( + self.executor.clone(), + request, + turn.into_response(), + )); } // A stall consumed by a simultaneous trigger does not latch, so the // checkpoint can still fire later if this review is refunded. @@ -370,7 +368,11 @@ impl AdvisorGate { self.mark_stall_fired(stall_key); } if !self.try_reserve(scope) { - return Ok(turn.into_response()); + return Ok(RoutingOutcome::answered( + self.executor.clone(), + request, + turn.into_response(), + )); } let trigger_label = match (&self.trigger, triggered) { @@ -385,11 +387,19 @@ impl AdvisorGate { .consult(driver, &request, review_tail.as_deref(), trigger_label) .await { - Ok(ConsultOutcome::Approve) => Ok(turn.into_response()), - Ok(ConsultOutcome::Redo { plan }) => self.redo(driver, request, turn, &plan).await, + Ok(ConsultOutcome::Approve) => Ok(RoutingOutcome::answered( + self.executor.clone(), + request, + turn.into_response(), + )), + Ok(ConsultOutcome::Redo { plan }) => Ok(self.redo(request, turn, &plan)), Ok(ConsultOutcome::Failed) => { self.refund_failure(scope); - Ok(turn.into_response()) + Ok(RoutingOutcome::answered( + self.executor.clone(), + request, + turn.into_response(), + )) } Err(error) => { self.refund_failure(scope); @@ -401,13 +411,7 @@ impl AdvisorGate { /// REDO: the client never sees the gated turn. Its text (or reasoning) is /// echoed as an assistant message, the advisor's plan follows as user /// feedback, and the executor continues as a pure passthrough call. - async fn redo( - &self, - driver: &Driver, - request: Request, - turn: GatedTurn, - plan: &str, - ) -> Result { + fn redo(&self, request: Request, turn: GatedTurn, plan: &str) -> RoutingOutcome { record_discarded(&turn.agg.usage); emit_discarded_audit(self.executor.as_str(), &turn.agg.usage); let echo = visible_text(&turn.agg) @@ -425,10 +429,7 @@ impl AdvisorGate { // preserved pre-surgery body verbatim and the feedback never reaches // the executor. crate::algorithms::util::prompts::drop_exact_replay(&mut redo); - driver.decide(self.executor_decision()).await?; - driver - .call_model(redo, vec![self.executor.clone()], true) - .await + RoutingOutcome::route_to(self.executor.clone(), Vec::new(), redo) } /// Consults the advisor over the buffered transcript and parses the @@ -462,10 +463,8 @@ impl AdvisorGate { ); let consult_request = self.build_consult_request(base, transcript); let started = Instant::now(); - // Judge-style call: the advisor never produces the client's answer, - // so no Decision is published for it. let reply = match driver - .call_model(consult_request, vec![self.advisor.clone()], false) + .call_model(consult_request, vec![self.advisor.clone()]) .await { Ok(response) => response @@ -584,7 +583,7 @@ impl Algorithm for AdvisorGate { "advisor_gate" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route(self: Arc, driver: Driver, request: Request) -> Result { let scope = budget_scope(&request); let session_final = request .metadata diff --git a/crates/libsy/src/algorithms/advisor_gate/tests.rs b/crates/libsy/src/algorithms/advisor_gate/tests.rs index 5d6f9b060..e57495668 100644 --- a/crates/libsy/src/algorithms/advisor_gate/tests.rs +++ b/crates/libsy/src/algorithms/advisor_gate/tests.rs @@ -10,7 +10,7 @@ use switchyard_protocol::{ResponseOutput, ToolCall, ToolResult, completion_text} use futures::StreamExt; use switchyard_protocol::{ AggLlmResponse, LlmClientError, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, ModelId, - StopReason, + Response, StopReason, }; use super::transcript::{NO_TEXT_PLACEHOLDER, TRUNCATION_MARKER, middle_drop}; @@ -270,7 +270,7 @@ async fn approved_terminal_turn_returns_buffered_body() { let script = Script::new(); let gate = gate(AdvisorGateConfig::default()); let serve = script.serve("APPROVE", |_| reply("all done")); - let (trace, response) = test_drive(gate, task_request(), serve) + let (selected_model, response) = test_drive(gate, task_request(), serve) .await .expect("routes"); assert_eq!( @@ -278,33 +278,7 @@ async fn approved_terminal_turn_returns_buffered_body() { vec![EXECUTOR.to_string(), ADVISOR.to_string()] ); assert_eq!(completion_text(&agg_of(response).await), "all done"); - // The published trace ends on the executor so hosts attribute the - // served model correctly. - let last = trace.last().expect("decision published"); - assert_eq!(last.selected_model_id(), EXECUTOR); - assert!(last.is_answer_call()); -} - -#[tokio::test] -async fn advisor_consult_publishes_no_decision() { - let script = Script::new(); - let gate = gate(AdvisorGateConfig::default()); - let serve = script.serve("APPROVE", |_| reply("done")); - let (trace, _) = test_drive(gate, task_request(), serve) - .await - .expect("routes"); - // The advisor was consulted... - assert_eq!( - script.models(), - vec![EXECUTOR.to_string(), ADVISOR.to_string()] - ); - // ...but as a judge-style call: no Decision is published for it, so - // hosts never attribute the served model to the advisor. - assert!(!trace.is_empty()); - for decision in &trace { - assert_eq!(decision.selected_model_id(), EXECUTOR); - assert!(decision.is_answer_call()); - } + assert_eq!(selected_model, EXECUTOR); } #[tokio::test] diff --git a/crates/libsy/src/algorithms/fall_through.rs b/crates/libsy/src/algorithms/fall_through.rs index be385a70e..4aa06c6b4 100644 --- a/crates/libsy/src/algorithms/fall_through.rs +++ b/crates/libsy/src/algorithms/fall_through.rs @@ -6,7 +6,7 @@ //! //! Each turn: request-side [`Processor`]s fold facts into the composition's state; the //! [`Classifier`] cascade is consulted in order and the first to score decides the target -//! (its `argmax`); the [`Decision`] is published and then replayed to the processors so +//! (its `argmax`); the selected model is replayed to the processors so //! stateful ones (latch, affinity) can bind it. //! //! The default `FallThrough<()>` carries no composition state. Stateful compositions share one @@ -29,8 +29,8 @@ use tokio::sync::Mutex as AsyncMutex; use crate::core::algorithm::{self, Algorithm, Driver}; use crate::core::classifier::{Classification, Classifier, Score}; use crate::core::processor::{Event, Processor}; -use crate::{LibsyError, Result}; -use switchyard_protocol::{Decision, ModelId, Request, Response}; +use crate::{LibsyError, Result, RoutingOutcome}; +use switchyard_protocol::{ModelId, Request, Response}; struct SessionState { state: Arc>, @@ -154,8 +154,8 @@ where self.classifiers.push(classifier); self } - /// Executes the processor/classifier/target-call sequence for wrappers and the trait entrypoint. - pub(crate) async fn execute(&self, driver: Driver, request: Request) -> Result { + /// Executes the processor and classifier sequence for wrappers and the trait entrypoint. + pub(crate) async fn execute(&self, driver: Driver, request: Request) -> Result { self.start_cleanup_task(); let session = session_id(&request); let session_final = request @@ -181,7 +181,7 @@ where }); } - async fn execute_session(&self, driver: Driver, request: Request) -> Result { + async fn execute_session(&self, driver: Driver, request: Request) -> Result { // The request is threaded mutably through the whole fold: any component may rewrite // it, later components see the rewrite, and the final value reaches the model. let mut request = request; @@ -203,11 +203,10 @@ where // Nothing reads it on the way out: streamed or buffered, it reaches the caller // untouched. match served { - Some(response) => Ok(response), + Some(response) => Ok(RoutingOutcome::answered(target, request, response)), None => { - driver - .call_model(request, self.candidates(&target), true) - .await + let fallback_models = self.fallbacks(&target); + Ok(RoutingOutcome::route_to(target, fallback_models, request)) } } } @@ -219,15 +218,12 @@ where } } - /// The selected target first, then every other configured target as a fallback candidate. - fn candidates(&self, target: &ModelId) -> Vec { - std::iter::once(target.clone()) - .chain( - self.targets - .iter() - .filter(|candidate| *candidate != target) - .cloned(), - ) + /// Every configured target other than the selection, in fallback order. + fn fallbacks(&self, target: &ModelId) -> Vec { + self.targets + .iter() + .filter(|candidate| *candidate != target) + .cloned() .collect() } @@ -274,21 +270,18 @@ where }); }; - // 3. Resolve the target, log the choice, and publish the decision. + // 3. Resolve the target and log the choice. algorithm::ensure_model_is_target(&self.targets, &score.target)?; let target = score.target.clone(); let message = (self.decision_reason)(&self.name, &score); let message = with_routing_tier(message, deciding.routing_tier(&target)); tracing::info!("{message}"); - let decision: Decision = Decision::new(target.clone(), true); - driver.decide(decision.clone()).await?; - - // 4. Post-decision replay: every processor sees the decision so stateful ones + // 4. Post-decision replay: every processor sees the selection so stateful ones // can bind it, and may rewrite the outbound request (e.g. add a target prompt). for processor in &self.processors { let event = Event::Decision { request, - decision: &decision, + selected_model_id: &target, }; processor.process(state, event).await?; } @@ -356,7 +349,7 @@ where &self.name } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route(self: Arc, driver: Driver, request: Request) -> Result { self.execute(driver, request).await } } @@ -533,30 +526,30 @@ mod tests { } } - /// Drives a shared router with one request, returning the completion text + trace. + /// Drives a shared router with one request, returning the completion text and selection. async fn run_request( router: &Arc>, request: Request, serve: impl Serve, - ) -> Result<(String, Vec)> + ) -> Result<(String, ModelId)> where S: Default + Send + 'static, { - let (trace, response) = test_drive(router.clone(), request, serve).await?; + let (selected_model, response) = test_drive(router.clone(), request, serve).await?; let text = response .llm_response .into_agg() .await .map(|agg| completion_text(&agg)) .map_err(|error| LibsyError::external("aggregating fall-through response", error))?; - Ok((text, trace)) + Ok((text, selected_model)) } /// Drives a shared router through one turn in the default test session. async fn run_turn( router: &Arc>, serve: impl Serve, - ) -> Result<(String, Vec)> + ) -> Result<(String, ModelId)> where S: Default + Send + 'static, { @@ -564,7 +557,7 @@ mod tests { } /// Drives a fresh router through one turn with a specific `serve`. - async fn run_with(router: FallThrough, serve: impl Serve) -> Result<(String, Vec)> { + async fn run_with(router: FallThrough, serve: impl Serve) -> Result<(String, ModelId)> { run_turn(&Arc::new(router), serve).await } @@ -667,23 +660,24 @@ mod tests { let stream = router.run_stream(request()); tokio::pin!(stream); while let Some(step) = stream.next().await { - if let crate::Step::CallModel(call) = step? { - assert_eq!(call.models, target_set(&["mid", "weak", "strong"])); - assert_eq!(call.request.llm_request.model.as_deref(), Some("mid")); + if let crate::Step::Done(outcome) = step? { + assert_eq!(outcome.selected_model_id, ModelId::from("mid")); + assert_eq!(outcome.fallback_models, target_set(&["weak", "strong"])); + assert_eq!(outcome.request.llm_request.model.as_deref(), Some("mid")); + assert!(outcome.response.is_none()); return Ok(()); } } - Err(test_error("expected a CallModel step")) + Err(test_error("expected a Done step")) } #[tokio::test] async fn argmax_picks_the_highest_confidence_target() -> Result<()> { let router = FallThrough::<()>::new(target_set(&["strong", "weak"])) .with_classifier(fixed(vec![score("weak", 0.2), score("strong", 0.9)])); - let (model, trace) = run_with(router, echo()).await?; + let (model, selected_model) = run_with(router, echo()).await?; assert_eq!(model, "strong"); - assert_eq!(trace.len(), 1); - assert_eq!(trace[0].selected_model_id(), "strong"); + assert_eq!(selected_model, "strong"); Ok(()) } diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index 8b146e039..589a20158 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -522,7 +522,7 @@ impl Classifier for EscalationClassifier { "escalation classifier selected efficient tier" ); let efficient_response = match driver - .call_model(request.clone(), vec![self.efficient.clone()], true) + .call_model(request.clone(), vec![self.efficient.clone()]) .await { Ok(r) => r, @@ -930,7 +930,11 @@ impl Algorithm for LlmTaskClassifier { "llm_task_classifier" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { self.route.execute(driver, request).await } } @@ -1124,12 +1128,10 @@ mod tests { async fn an_unreachable_judge_routes_capable_instead_of_failing_the_request() -> Result<()> { let router = router()?; - let (trace, response) = test_drive(router, classify_request(), unreachable_judge()).await?; + let (selected_model, response) = + test_drive(router, classify_request(), unreachable_judge()).await?; - assert_eq!( - trace.last().map(|d| d.selected_model_id().as_str()), - Some("capable") - ); + assert_eq!(selected_model, "capable"); assert_eq!( response.llm_response.as_agg().map(completion_text), Some("answer from capable".to_string()) @@ -1814,14 +1816,11 @@ mod tests { let model = Queue::new(["efficient answer"]); let router = escalation_router()?; - let (trace, response) = + let (selected_model, response) = test_drive(router, classify_request(), queued(model, judge)).await?; // The efficient model is the serving target, and the response comes from its call. - assert_eq!( - trace.last().map(|d| d.selected_model_id().as_str()), - Some("efficient") - ); + assert_eq!(selected_model, "efficient"); assert_eq!( response.llm_response.as_agg().map(completion_text), Some("efficient answer".to_string()) @@ -1860,13 +1859,10 @@ mod tests { let model = Queue::new(["efficient draft", "capable answer"]); let router = escalation_router()?; - let (trace, response) = + let (selected_model, response) = test_drive(router, classify_request(), queued(model, judge)).await?; - assert_eq!( - trace.last().map(|d| d.selected_model_id().as_str()), - Some("capable") - ); + assert_eq!(selected_model, "capable"); assert_eq!( response.llm_response.as_agg().map(completion_text), Some("capable answer".to_string()) @@ -1889,12 +1885,10 @@ mod tests { queued(Arc::clone(&model), Arc::clone(&judge)), ) .await?; - let (trace, _) = test_drive(router.clone(), session_request, queued(model, judge)).await?; + let (selected_model, _) = + test_drive(router.clone(), session_request, queued(model, judge)).await?; - assert_eq!( - trace.last().map(|d| d.selected_model_id().as_str()), - Some("capable") - ); + assert_eq!(selected_model, "capable"); Ok(()) } @@ -1917,12 +1911,9 @@ mod tests { } }; - let (trace, response) = test_drive(router, classify_request(), serve).await?; + let (selected_model, response) = test_drive(router, classify_request(), serve).await?; - assert_eq!( - trace.last().map(|d| d.selected_model_id().as_str()), - Some("capable") - ); + assert_eq!(selected_model, "capable"); assert_eq!( response.llm_response.as_agg().map(completion_text), Some("capable answer".to_string()) diff --git a/crates/libsy/src/algorithms/noop.rs b/crates/libsy/src/algorithms/noop.rs index 90e053b7a..95f4c7212 100644 --- a/crates/libsy/src/algorithms/noop.rs +++ b/crates/libsy/src/algorithms/noop.rs @@ -10,9 +10,8 @@ use switchyard_protocol::{ StopReason, }; -use crate::Result; use crate::core::algorithm::{Algorithm, Driver}; -use switchyard_protocol::Decision; +use crate::{Result, RoutingOutcome}; /// Test helper that returns a hard-coded response without routing or model I/O. pub struct Noop {} @@ -23,14 +22,11 @@ impl Algorithm for Noop { "noop" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route(self: Arc, _driver: Driver, request: Request) -> Result { let model_id = request .model_id() .unwrap_or_else(|| ModelId::from("switchyard/noop")); tracing::info!(target = %model_id, "noop returned its synthetic response"); - let decision: Decision = Decision::new(model_id.clone(), true); - driver.decide(decision.clone()).await?; - let llm_response = LlmResponse::Agg(AggLlmResponse { id: Some("switchyard-noop".to_string()), model: Some(model_id.to_string()), @@ -47,7 +43,7 @@ impl Algorithm for Noop { llm_response, metadata: request.metadata.clone(), }; - Ok(response) + Ok(RoutingOutcome::answered(model_id, request, response)) } } @@ -74,12 +70,8 @@ mod tests { // `Noop` synthesizes its own response and never offloads a call, so `echo` is // never reached. let a: Arc = Arc::new(Noop {}); - let (decisions, response) = test_drive(a, request, echo()).await?; - let Some(decision) = decisions.first() else { - panic!("Expected exactly one Decision"); - }; - assert_eq!(decision.selected_model_id(), TEST_MODEL); - assert!(decision.is_answer_call()); + let (selected_model, response) = test_drive(a, request, echo()).await?; + assert_eq!(selected_model, TEST_MODEL); assert_eq!(response.selected_model(), Some(TEST_MODEL)); Ok(()) } diff --git a/crates/libsy/src/algorithms/passthrough.rs b/crates/libsy/src/algorithms/passthrough.rs index 3de5ccc9a..0ea325bf9 100644 --- a/crates/libsy/src/algorithms/passthrough.rs +++ b/crates/libsy/src/algorithms/passthrough.rs @@ -5,19 +5,18 @@ use std::sync::Arc; -use switchyard_protocol::{ModelId, Request, Response}; +use switchyard_protocol::{ModelId, Request}; -use crate::Result; use crate::core::algorithm::{Algorithm, Driver}; -use switchyard_protocol::Decision; +use crate::{Result, RoutingOutcome}; -/// Routing algorithm that always calls one configured target. +/// Routing algorithm that always selects one configured target. pub struct Passthrough { target: ModelId, } impl Passthrough { - /// Creates an algorithm that always calls `target`. + /// Creates an algorithm that always selects `target`. pub fn new(target: impl Into) -> Self { Passthrough { target: target.into(), @@ -31,13 +30,13 @@ impl Algorithm for Passthrough { "passthrough" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route(self: Arc, _driver: Driver, request: Request) -> Result { tracing::info!(target = %self.target, "passthrough selected target"); - let decision: Decision = Decision::new(self.target.clone(), true); - driver.decide(decision.clone()).await?; - driver - .call_model(request, vec![self.target.clone()], true) - .await + Ok(RoutingOutcome::route_to( + self.target.clone(), + Vec::new(), + request, + )) } } @@ -59,7 +58,7 @@ mod tests { metadata: None, }; let algorithm: Arc = Arc::new(Passthrough::new(MODEL_ID)); - let (trace, response) = test_drive(algorithm, request, echo()).await?; + let (selected_model, response) = test_drive(algorithm, request, echo()).await?; assert_eq!( response @@ -69,9 +68,7 @@ mod tests { .unwrap_or_default(), MODEL_ID ); - assert_eq!(trace.len(), 1); - assert_eq!(trace[0].selected_model_id(), MODEL_ID); - assert!(trace[0].is_answer_call()); + assert_eq!(selected_model, MODEL_ID); Ok(()) } } diff --git a/crates/libsy/src/algorithms/rand.rs b/crates/libsy/src/algorithms/rand.rs index 4be406f32..e3e04b034 100644 --- a/crates/libsy/src/algorithms/rand.rs +++ b/crates/libsy/src/algorithms/rand.rs @@ -158,7 +158,11 @@ impl Algorithm for Random { "random" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { self.inner.execute(driver, request).await } } @@ -222,7 +226,7 @@ mod tests { #[tokio::test] async fn single_target_is_always_selected_and_called() -> Result<()> { let algorithm = shared_algorithm(&["only/model"])?; - let (trace, response) = test_drive(algorithm, request(), echo()).await?; + let (selected_model, response) = test_drive(algorithm, request(), echo()).await?; assert_eq!( response @@ -232,29 +236,7 @@ mod tests { .unwrap_or_default(), "only/model" ); - assert_eq!(trace.len(), 1); - assert_eq!(trace[0].selected_model_id(), "only/model"); - Ok(()) - } - - #[tokio::test] - async fn selected_target_is_in_the_set_and_matches_the_trace() -> Result<()> { - let names = ["a/model", "b/model", "c/model"]; - let algorithm = shared_algorithm(&names)?; - - for _ in 0..50 { - let (trace, response) = test_drive(algorithm.clone(), request(), echo()).await?; - let selected = response - .llm_response - .as_agg() - .map(completion_text) - .unwrap_or_default(); - assert!( - names.contains(&selected.as_str()), - "selected {selected} not in target set" - ); - assert_eq!(trace[0].selected_model_id(), selected.as_str()); - } + assert_eq!(selected_model, "only/model"); Ok(()) } @@ -264,14 +246,15 @@ mod tests { let mut seen = HashSet::new(); for _ in 0..100 { - let (_, response) = test_drive(algorithm.clone(), request(), echo()).await?; - seen.insert( - response - .llm_response - .as_agg() - .map(completion_text) - .unwrap_or_default(), - ); + let (selected_model, response) = + test_drive(algorithm.clone(), request(), echo()).await?; + let served_model = response + .llm_response + .as_agg() + .map(completion_text) + .unwrap_or_default(); + assert_eq!(selected_model, served_model.as_str()); + seen.insert(served_model); } // Missing either target after 100 uniform draws has probability about 2^-99. @@ -393,11 +376,8 @@ mod tests { #[tokio::test] async fn decision_is_inspectable() -> Result<()> { let algorithm = shared_algorithm(&["only/model"])?; - let (trace, _) = test_drive(algorithm, request(), echo()).await?; - let decision = &trace[0]; - - assert_eq!(decision.selected_model_id(), "only/model"); - assert!(decision.is_answer_call()); + let (selected_model, _) = test_drive(algorithm, request(), echo()).await?; + assert_eq!(selected_model, "only/model"); Ok(()) } } diff --git a/crates/libsy/src/algorithms/stage.rs b/crates/libsy/src/algorithms/stage.rs index f751f7939..00ec3e59e 100644 --- a/crates/libsy/src/algorithms/stage.rs +++ b/crates/libsy/src/algorithms/stage.rs @@ -139,7 +139,11 @@ impl Algorithm for StageRouter { STAGE_ROUTER } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { self.route.execute(driver, request).await } } @@ -334,7 +338,6 @@ mod tests { struct Call { target: String, messages: Vec, - is_answer_call: bool, } /// Records what each target receives. @@ -349,7 +352,7 @@ mod tests { self.calls .lock() .iter() - .filter(|call| call.is_answer_call) + .filter(|call| call.target != JUDGE) .cloned() .collect() } @@ -370,7 +373,6 @@ mod tests { .iter() .filter_map(|message| message.text_content("|")) .collect(), - is_answer_call: target != JUDGE, }); let completion = if target == JUDGE { let p_solve = *recorder.judge_p_solve.lock(); @@ -495,28 +497,20 @@ mod tests { let recorder = Arc::new(Recorder::default()); let router = recording_router(config_with_judge(&recorder, 0.1))?; - let (trace, _) = test_drive(router.clone(), turn_request(false), recorder.serve()).await?; + let (selected_model, _) = + test_drive(router.clone(), turn_request(false), recorder.serve()).await?; let calls = recorder.calls.lock(); assert!( - calls - .iter() - .any(|call| call.target == JUDGE && !call.is_answer_call), + calls.iter().any(|call| call.target == JUDGE), "the judge should be recorded as a routing side call" ); assert!( - calls - .iter() - .any(|call| call.target == "strong" && call.is_answer_call), + calls.iter().any(|call| call.target == "strong"), "the selected target should be recorded as an answer call" ); drop(calls); - assert_eq!( - trace - .last() - .map(|decision| decision.selected_model_id().as_str()), - Some("strong") - ); + assert_eq!(selected_model, "strong"); Ok(()) } diff --git a/crates/libsy/src/algorithms/util/affinity.rs b/crates/libsy/src/algorithms/util/affinity.rs index 5f5db53e7..5efe21d7c 100644 --- a/crates/libsy/src/algorithms/util/affinity.rs +++ b/crates/libsy/src/algorithms/util/affinity.rs @@ -146,14 +146,16 @@ where S: Send + 'static, { async fn process(&self, _state: &mut S, event: Event<'_>) -> crate::Result<()> { - if let Event::Decision { request, decision } = event + if let Event::Decision { + request, + selected_model_id, + } = event && let Some(key) = self.affinity_key(request) { - let model = decision.selected_model_id(); let mut assignments = self.assignments.lock(); - if self.should_latch(model) && !assignments.contains_key(&key) { + if self.should_latch(selected_model_id) && !assignments.contains_key(&key) { evict_if_full(&mut assignments); - assignments.insert(key, model.clone()); + assignments.insert(key, selected_model_id.clone()); } } Ok(()) @@ -217,15 +219,13 @@ mod tests { use std::sync::Arc; - use switchyard_protocol::{ - ContentBlock, Decision, LlmRequest, Message, Metadata, text_request, - }; + use switchyard_protocol::{ContentBlock, LlmRequest, Message, Metadata, text_request}; /// Boxed, thread-safe error type keeping the test helpers ergonomic. type BoxErr = Box; - fn fixed_decision(target: &str) -> Decision { - Decision::new(target, true) + fn fixed_model(target: &str) -> ModelId { + ModelId::from(target) } fn request(metadata: Metadata) -> Request { @@ -285,13 +285,13 @@ mod tests { request: &mut Request, model: &'static str, ) -> Result<(), BoxErr> { - let decision = fixed_decision(model); + let selected_model_id = ModelId::from(model); router .process( state, Event::Decision { request, - decision: &decision, + selected_model_id: &selected_model_id, }, ) .await?; @@ -584,7 +584,7 @@ mod tests { &mut state, Event::Decision { request: &mut first, - decision: &fixed_decision("model-a"), + selected_model_id: &fixed_model("model-a"), }, ) .await?; @@ -609,7 +609,7 @@ mod tests { &mut state, Event::Decision { request: &mut unkeyed, - decision: &fixed_decision("model-a"), + selected_model_id: &fixed_model("model-a"), }, ) .await?; @@ -633,7 +633,7 @@ mod tests { &mut state, Event::Decision { request: &mut second, - decision: &fixed_decision("model-b"), + selected_model_id: &fixed_model("model-b"), }, ) .await?; @@ -642,7 +642,7 @@ mod tests { &mut state, Event::Decision { request: &mut first, - decision: &fixed_decision("model-a"), + selected_model_id: &fixed_model("model-a"), }, ) .await?; diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs index b9fad1375..fb2dca589 100644 --- a/crates/libsy/src/algorithms/util/llm_judge.rs +++ b/crates/libsy/src/algorithms/util/llm_judge.rs @@ -241,7 +241,6 @@ where .call_model( self.judge.build_request(state, request), vec![self.target.clone()], - false, ) .await .inspect_err(|error| report_fail_open(judge_model, error, libsy_error_reason(error))) diff --git a/crates/libsy/src/algorithms/util/prompts.rs b/crates/libsy/src/algorithms/util/prompts.rs index 8118b3472..37314d43c 100644 --- a/crates/libsy/src/algorithms/util/prompts.rs +++ b/crates/libsy/src/algorithms/util/prompts.rs @@ -114,10 +114,14 @@ impl Processor for SystemPromptProcessor { // The decision event carries both the routing outcome and the outbound request, // so the target is read straight off it — whichever classifier picked it, and // with nothing kept between turns. - let Event::Decision { request, decision } = event else { + let Event::Decision { + request, + selected_model_id, + } = event + else { return Ok(()); }; - let Some(prompt) = self.prompts.get(decision.selected_model_id()) else { + let Some(prompt) = self.prompts.get(selected_model_id) else { return Ok(()); }; // Ahead of the client's own instructions, so this framing is what the @@ -139,7 +143,7 @@ impl Processor for SystemPromptProcessor { #[cfg(test)] mod tests { use super::*; - use switchyard_protocol::{Decision, LlmRequest, ToolResult, text_request}; + use switchyard_protocol::{LlmRequest, ModelId, ToolResult, text_request}; const NOTE: &str = "recovering from an error"; const STRONG_PROMPT: &str = "diagnose before you edit"; @@ -265,13 +269,13 @@ mod tests { }, ..Request::default() }; - let decision = Decision::new(target, true); + let selected_model_id = ModelId::from(target); processor .process( &mut (), Event::Decision { request: &mut request, - decision: &decision, + selected_model_id: &selected_model_id, }, ) .await?; @@ -326,14 +330,14 @@ mod tests { text: "you are a coding agent".to_string(), }], }); - let decision = Decision::new("strong", true); + let selected_model_id = ModelId::from("strong"); processor .process( &mut (), Event::Decision { request: &mut request, - decision: &decision, + selected_model_id: &selected_model_id, }, ) .await?; diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index f46b4815e..47ddfb668 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -2,8 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 //! The [`Algorithm`] trait and its [`Driver`] — the orchestration contract every -//! algorithm implements, and the offload channel it uses to make model calls and -//! publish [`Decision`]s. +//! algorithm implements and the offload channel it uses for routing-time model calls. use std::{future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc, time::Instant}; @@ -20,7 +19,7 @@ use tracing::Instrument; /// [`switchyard_protocol::LlmResponseStreamEvent`] is its host/algorithm envelope; and /// [`switchyard_protocol::LlmResponse`] carries either a live /// [`switchyard_protocol::LlmResponseStream`] or the terminal aggregate. -use switchyard_protocol::{Decision, ModelId, Request, Response}; +use switchyard_protocol::{ModelId, Request, Response}; use crate::{DriverError, LibsyError, Result, observability}; @@ -34,8 +33,7 @@ pub type StepStream = Pin> + Send>>; /// The host reads the public fields, performs (or delegates) the model call, and fulfills it /// with [`respond`](Self::respond) — unblocking the algorithm's [`Driver::call_model`] on the /// other side. `switchyard-llm-client`'s `run` is the ready-made consumer that does this for -/// you. A host that only wants the routing outcome can take the contents with -/// [`into_parts`](Self::into_parts) and never respond; dropping the stream ends the run. +/// you. /// /// [`Driver::call_model`] stamps the first candidate model onto the request before publishing /// the call. A consumer that falls through to a later candidate must re-stamp it. @@ -47,8 +45,6 @@ pub struct CallModel { pub request: Request, /// Candidate models, tried in order until one answers. Never empty. pub models: Vec, - /// True for an answer-generating call, false for classifier and judge calls. - pub is_answer_call: bool, // How to send the response back to the algorithm reply: oneshot::Sender>, } @@ -62,29 +58,48 @@ impl CallModel { .send(result) .map_err(|_| DriverError::ResponseDropped.into()) } +} - /// Take the call's contents without answering it, dropping the promise — the routing - /// outcome plus the request as the algorithm would have sent it, after any rewriting. - /// - /// Should only be called if `is_answer_call` is true as that is the final call. - /// The algorithm's [`Driver::call_model`] will fail with [`DriverError::Abandoned`] and - /// the run ends there. Taking a call the algorithm does not depend on (a judge or - /// classifier call) may instead let it fail open and complete with degraded routing. - /// - /// An abandoned run is not recorded as a failed one. Dropping a [`CallModel`] without - /// calling this still yields [`DriverError::ResponseDropped`], which is. - pub fn into_parts(self) -> (Request, Vec) { - let Self { +/// The terminal result of routing. +pub struct RoutingOutcome { + /// The model selected by the algorithm and tried first by the client. + pub selected_model_id: ModelId, + /// Additional models the client may try in order after an eligible failure. + pub fallback_models: Vec, + /// The request after all routing-time rewrites, stamped with the selected model. + pub request: Request, + /// A response produced while routing, or `None` when the client must make the answer call. + pub response: Option, +} + +impl RoutingOutcome { + /// The decision is that client should send this `request`. The `selected_model_id` + /// will be written into it by this function. + /// If that fails client should try the `fallback_models` in order. + pub fn route_to( + selected_model_id: ModelId, + fallback_models: Vec, + mut request: Request, + ) -> Self { + request.llm_request.model = Some(selected_model_id.to_string()); + Self { + selected_model_id, + fallback_models, request, - models, - reply, - .. - } = self; - // Tell the algorithm the call was taken deliberately rather than lost, so its - // telemetry can tell an abandoned run from a failed one. The receiver is already - // gone if the algorithm stopped waiting, which is fine. - let _ = reply.send(Err(DriverError::Abandoned.into())); - (request, models) + response: None, + } + } + + /// Algorithm generated the response as part of the routing decision. Here it is. + /// The `request` will have the `selected_model_id` written into it by this function. + pub fn answered(selected_model_id: ModelId, mut request: Request, response: Response) -> Self { + request.llm_request.model = Some(selected_model_id.to_string()); + Self { + selected_model_id, + fallback_models: Vec::new(), + request, + response: Some(response), + } } } @@ -92,8 +107,7 @@ impl CallModel { #[derive(Clone)] pub struct Driver { step_tx: mpsc::Sender>, - /// The owning algorithm's telemetry label, stamped onto every call and decision - /// this driver publishes. + /// The owning algorithm's telemetry label, stamped onto every call this driver publishes. algorithm: String, } @@ -139,12 +153,7 @@ impl Driver { reasoning_tokens = tracing::field::Empty, ) )] - pub async fn call_model( - &self, - mut request: Request, - models: Vec, - is_answer_call: bool, - ) -> Result { + pub async fn call_model(&self, mut request: Request, models: Vec) -> Result { let Some(selected_model_id) = models.first().cloned() else { return Err(LibsyError::NoTargets); }; @@ -155,7 +164,6 @@ impl Driver { algorithm: self.algorithm.clone(), request, models, - is_answer_call, reply, }; let result = async { @@ -172,7 +180,6 @@ impl Driver { observability::record_llm_call( &self.algorithm, selected_model_id.as_str(), - is_answer_call, elapsed, &result, &tracing::Span::current(), @@ -180,27 +187,23 @@ impl Driver { result } - /// Publish a routing [`Decision`] as a [`Step::Decision`] on the stream. - /// Each successfully published decision is counted and logged; a decision - /// the stream never accepted is not recorded. - pub async fn decide(&self, decision: Decision) -> Result<()> { - self.step_tx - .send(Ok(Step::Decision(decision.clone()))) - .await - .map_err(|_| DriverError::StreamClosed)?; - observability::record_decision(&self.algorithm, &decision); - Ok(()) - } - /// Emit the terminal step: [`Step::Done`] on `Ok`, or an `Err` stream /// item on failure. Internal: called once by [`run_stream`](Algorithm::run_stream) /// when the algorithm finishes. - pub(crate) async fn finish(&self, result: Result) -> Result<()> { - let step = result.map(|response| Step::Done(Box::new(response))); + pub(crate) async fn finish(&self, result: Result) -> Result<()> { + let selected_model = result + .as_ref() + .ok() + .map(|outcome| outcome.selected_model_id.clone()); + let step = result.map(|outcome| Step::Done(Box::new(outcome))); self.step_tx .send(step) .await - .map_err(|_| DriverError::StreamClosed.into()) + .map_err(|_| DriverError::StreamClosed)?; + if let Some(selected_model) = selected_model { + observability::record_decision(&self.algorithm, &selected_model); + } + Ok(()) } } @@ -209,16 +212,13 @@ pub enum Step { /// The algorithm needs this model call performed. The host serves it and fulfills /// it with [`CallModel::respond`]. Boxed: it is by far the largest variant. CallModel(Box), - /// A routing decision the algorithm made, published via [`Driver::decide`] as it - /// happens (rather than collected into a trace returned at the end). - Decision(Decision), - /// The algorithm finished with its final response — the last step of a run. - Done(Box), + /// The algorithm finished with its routing outcome — the last step of a run. + Done(Box), } /// Drive [`Algorithm::run_stream`] to completion, handing each offloaded call to `serve`. /// -/// Returns the final [`Response`] and the trace of [`Decision`]s the algorithm published. +/// Returns the final [`RoutingOutcome`]. /// `serve` owns the call: it performs it however the host likes and must fulfill the promise /// with [`CallModel::respond`]. A failed *model* call belongs in `respond` — the /// algorithm may route around it. Returning `Err` from `serve` aborts the whole run, so @@ -232,7 +232,7 @@ pub async fn drive( algorithm: Arc, request: Request, serve: F, -) -> Result<(Vec, Response)> +) -> Result where F: Fn(CallModel) -> Fut, Fut: Future>, @@ -240,9 +240,8 @@ where let stream = algorithm.run_stream(request); tokio::pin!(stream); - let mut trace: Vec = Vec::new(); let mut in_flight = futures::stream::FuturesUnordered::new(); - let mut final_response: Option = None; + let mut final_outcome: Option = None; loop { tokio::select! { @@ -255,9 +254,8 @@ where None => break, // stream has ended, no more steps Some(item) => match item? { Step::CallModel(call) => in_flight.push(serve(*call)), - Step::Decision(decision) => trace.push(decision), - Step::Done(response) => { - final_response = Some(*response); + Step::Done(outcome) => { + final_outcome = Some(*outcome); break; } } @@ -265,9 +263,7 @@ where }, } } - final_response - .map(|response| (trace, response)) - .ok_or(LibsyError::MissingFinalResponse) + final_outcome.ok_or(LibsyError::MissingFinalResponse) } /// Recover the message from an algorithm's panic. @@ -350,7 +346,7 @@ impl RoutingIdentity { /// # Observability /// /// [`run_stream`](Self::run_stream) creates a `libsy.run` span, and each offloaded model -/// call creates a `libsy.llm_call` span. Decisions and failures are emitted through +/// call creates a `libsy.llm_call` span. Routing decisions and failures are emitted through /// `tracing`; metrics use the global OpenTelemetry meter provider. The provider call /// itself belongs to the host, and is instrumented by whoever makes it. #[async_trait] @@ -360,10 +356,10 @@ pub trait Algorithm: Send + Sync + 'static { /// emits for its runs. fn name(&self) -> &str; - /// Run one request to completion: make model calls with [`Driver::call_model`], - /// publish [`Decision`]s with [`Driver::decide`], and return the final [`Response`]. + /// Run one request to completion: make routing-time model calls with + /// [`Driver::call_model`] and return the terminal [`RoutingOutcome`]. /// The method an algorithm implements; [`run_stream`](Self::run_stream) drives it. - async fn route(self: Arc, driver: Driver, request: Request) -> Result; + async fn route(self: Arc, driver: Driver, request: Request) -> Result; /// Process a request to completion, returning a stream of [`Step`]s. /// @@ -426,13 +422,8 @@ mod tests { LibsyError::external("test", TestError(message)) } - /// Build a routed decision for orchestration tests. - fn test_decision(selected_model_id: ModelId) -> Decision { - Decision::new(selected_model_id, true) - } - /// Trivial algo used only to exercise the orchestrator: calls the first target - /// and returns its response with a one-item trace. + /// and returns its response as the routing outcome. struct TestAlgo { target_set: Vec, } @@ -443,15 +434,20 @@ mod tests { "test" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { let target = self .target_set .first() .ok_or(LibsyError::NoTargets)? .clone(); - let decision = test_decision(target.clone()); - driver.decide(decision.clone()).await?; - driver.call_model(request, vec![target], true).await + let response = driver + .call_model(request.clone(), vec![target.clone()]) + .await?; + Ok(RoutingOutcome::answered(target, request, response)) } } @@ -468,6 +464,47 @@ mod tests { } } + #[test] + fn routing_outcome_constructors_stamp_selection_and_preserve_payloads() { + let outcome = RoutingOutcome::route_to( + "selected".into(), + target_set(&["fallback-one", "fallback-two"]), + request(), + ); + + assert_eq!(outcome.selected_model_id, "selected"); + assert_eq!( + outcome.fallback_models, + target_set(&["fallback-one", "fallback-two"]) + ); + assert_eq!(outcome.request.model_id().as_deref(), Some("selected")); + assert!(outcome.response.is_none()); + + let outcome = RoutingOutcome::route_to("only".into(), Vec::new(), request()); + assert!(outcome.fallback_models.is_empty()); + + let outcome = RoutingOutcome::answered( + "answered".into(), + request(), + Response { + llm_response: LlmResponse::Agg(text_response(None, "existing")), + metadata: None, + }, + ); + + assert_eq!(outcome.selected_model_id, "answered"); + assert_eq!(outcome.request.model_id().as_deref(), Some("answered")); + assert!(outcome.fallback_models.is_empty()); + assert_eq!( + outcome + .response + .as_ref() + .and_then(|response| response.llm_response.as_agg()) + .map(completion_text), + Some("existing".to_string()) + ); + } + fn target_set(names: &[&str]) -> Vec { names.iter().map(|name| ModelId::from(*name)).collect() } @@ -481,12 +518,12 @@ mod tests { let first_driver = driver.clone(); let mut first = tokio::spawn(async move { first_driver - .call_model(request(), vec![ModelId::from("first")], true) + .call_model(request(), vec![ModelId::from("first")]) .await }); let second = tokio::spawn(async move { driver - .call_model(request(), vec![ModelId::from("second")], true) + .call_model(request(), vec![ModelId::from("second")]) .await }); @@ -537,7 +574,7 @@ mod tests { let (driver, mut step_rx) = Driver::new("test"); let producer = tokio::spawn(async move { driver - .call_model(request(), vec![ModelId::from("dropped")], true) + .call_model(request(), vec![ModelId::from("dropped")]) .await }); let step = step_rx.recv().await.ok_or(DriverError::StreamClosed)??; @@ -553,11 +590,12 @@ mod tests { Err(LibsyError::Driver(DriverError::ResponseDropped)) )); - // A standalone driver reports the typed step receiver disappearing at its next send. + // A standalone driver reports the typed step receiver disappearing at its next call. let (driver, step_rx) = Driver::new("test"); drop(step_rx); - let decision = test_decision(ModelId::from("closed")); - let result = driver.decide(decision).await; + let result = driver + .call_model(request(), vec![ModelId::from("closed")]) + .await; assert!(matches!( result, Err(LibsyError::Driver(DriverError::StreamClosed)) @@ -568,55 +606,6 @@ mod tests { .map_err(|error| LibsyError::external("waiting for typed driver boundaries", error))? } - /// A consumer that only wants the routing outcome takes the call apart instead of - /// answering it: it gets the request as the algorithm would have sent it, and the - /// algorithm learns the call was abandoned rather than lost — the distinction that - /// keeps a deliberate decision-only run out of the failure counters. - #[tokio::test] - async fn into_parts_yields_the_selected_model_without_answering_it() -> Result<()> { - let (driver, mut step_rx) = Driver::new("test"); - let producer = tokio::spawn(async move { - driver - .call_model(request(), vec![ModelId::from("answer/model")], true) - .await - }); - - let step = step_rx.recv().await.ok_or(DriverError::StreamClosed)??; - let Step::CallModel(call) = step else { - return Err(test_error("expected a CallModel step")); - }; - let (taken_request, taken_models) = call.into_parts(); - assert_eq!(taken_models, vec![ModelId::from("answer/model")]); - assert_eq!( - taken_request.llm_request.model.as_deref(), - Some("answer/model") - ); - let mut expected = request().llm_request; - expected.model = Some("answer/model".to_string()); - assert_eq!(taken_request.llm_request, expected); - - let result = producer - .await - .map_err(|source| LibsyError::external("joining a test task", source))?; - assert!(matches!( - result, - Err(LibsyError::Driver(DriverError::Abandoned)) - )); - Ok(()) - } - - /// The two ways a call goes unanswered are told apart: `into_parts` is the consumer's - /// choice, a bare drop is the promise being lost. - #[test] - fn abandoning_and_dropping_a_call_report_different_outcomes() { - let abandoned: Result = Err(DriverError::Abandoned.into()); - let dropped: Result = Err(DriverError::ResponseDropped.into()); - assert_eq!(observability::outcome_value(&abandoned), "abandoned"); - assert_eq!(observability::outcome_value(&dropped), "error"); - assert!(observability::is_abandoned(&abandoned)); - assert!(!observability::is_abandoned(&dropped)); - } - #[test] fn target_lookup_returns_the_missing_target() { let error = ensure_model_is_target(&target_set(&[]), &ModelId::from("missing")).err(); @@ -665,7 +654,7 @@ mod tests { reason: Some("stop".to_string()), }, ]); - let (trace, response) = test_drive(orch, request(), serve).await?; + let (selected_model, response) = test_drive(orch, request(), serve).await?; // The run handed back the live stream; the caller folds it to a buffered aggregate. let agg = response .llm_response @@ -674,7 +663,7 @@ mod tests { .map_err(|error| LibsyError::external("aggregating response stream", error))?; assert_eq!(completion_text(&agg), "hello"); assert_eq!(agg.model.as_deref(), Some("stream/model")); - assert_eq!(trace.len(), 1); + assert_eq!(selected_model, "stream/model"); Ok(()) } @@ -724,10 +713,10 @@ mod tests { metadata: None, }))?; } - Step::Decision(decision) => { - assert_eq!(decision.selected_model_id(), "offload/model"); - } - Step::Done(response) => { + Step::Done(outcome) => { + let response = outcome + .response + .ok_or_else(|| test_error("expected an answered outcome"))?; final_completion = Some( response .llm_response @@ -747,23 +736,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn a_driven_run_returns_the_trace_and_the_final_response() -> Result<()> { - let (trace, response) = - test_drive(orch(target_set(&["direct/model"])), request(), echo()).await?; - // TestAlgo calls the first target; `echo` answers with its name. - assert_eq!( - response - .llm_response - .as_agg() - .map(completion_text) - .unwrap_or_default(), - "direct/model" - ); - assert_eq!(trace[0].selected_model_id(), "direct/model"); - Ok(()) - } - #[tokio::test(flavor = "multi_thread", worker_threads = 12)] async fn requests_are_processed_in_parallel() -> Result<()> { use std::time::Duration; @@ -828,7 +800,6 @@ mod tests { Ok(Step::CallModel(call)) => { call.respond(Err(test_error("upstream model call failed")))?; } - Ok(Step::Decision(_)) => {} Ok(Step::Done(..)) => { return Err(test_error( "expected the offload error to propagate, got a response", @@ -876,7 +847,7 @@ mod tests { self: Arc, _driver: Driver, _request: Request, - ) -> Result { + ) -> Result { let _guard = DropGuard(self.dropped.clone()); let _ = self.started.send(()); // Await forever without ever touching the driver. @@ -923,7 +894,7 @@ mod tests { self: Arc, _driver: Driver, _request: Request, - ) -> Result { + ) -> Result { panic!("boom"); } } @@ -961,7 +932,11 @@ mod tests { "leaky_panic" } - async fn route(self: Arc, driver: Driver, _request: Request) -> Result { + async fn route( + self: Arc, + driver: Driver, + _request: Request, + ) -> Result { tokio::spawn(async move { // Outlives the panic below, keeping a sender clone alive. let _keep_alive = driver; @@ -1022,7 +997,7 @@ mod tests { self: Arc, _driver: Driver, _request: Request, - ) -> Result { + ) -> Result { let _guard = DropGuard(self.dropped.clone()); let _ = self.started.send(()); // Hang forever without ever touching the driver, so only cancellation @@ -1071,13 +1046,26 @@ mod tests { "hedge" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { - let win = driver.call_model(request.clone(), vec![self.winner.clone().into()], true); - let lose = driver.call_model(request, vec![self.loser.clone().into()], true); + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { + let outcome_request = request.clone(); + let win = driver.call_model(request.clone(), vec![self.winner.clone().into()]); + let lose = driver.call_model(request, vec![self.loser.clone().into()]); // First to resolve wins; `select!` drops the losing future (and its promise). tokio::select! { - res = win => res, - res = lose => res, + res = win => Ok(RoutingOutcome::answered( + self.winner.clone().into(), + outcome_request, + res?, + )), + res = lose => Ok(RoutingOutcome::answered( + self.loser.clone().into(), + outcome_request, + res?, + )), } } } @@ -1114,7 +1102,7 @@ mod tests { // The loser responds 50ms after the winner has already won. `run` must return the // winner, not the loser's `respond`-to-a-dropped-receiver error. let (algo, serve) = hedge(Some(std::time::Duration::from_millis(50))); - let (_trace, response) = test_drive(algo, request(), serve).await?; + let (_, response) = test_drive(algo, request(), serve).await?; assert_eq!( response .llm_response @@ -1132,7 +1120,7 @@ mod tests { // waiting for the in-flight loser. let (algo, serve) = hedge(None); let run = test_drive(algo, request(), serve); - let (_trace, response) = tokio::time::timeout(std::time::Duration::from_secs(1), run) + let (_, response) = tokio::time::timeout(std::time::Duration::from_secs(1), run) .await .map_err(|error| LibsyError::external("waiting for pending loser", error))??; assert_eq!( @@ -1167,10 +1155,15 @@ mod tests { "fan_out_then_error" } - async fn route(self: Arc, driver: Driver, request: Request) -> Result { - let offloads = futures::future::join_all((0..self.n).map(|i| { - driver.call_model(request.clone(), vec![format!("m{i}").into()], true) - })); + async fn route( + self: Arc, + driver: Driver, + request: Request, + ) -> Result { + let offloads = futures::future::join_all( + (0..self.n) + .map(|i| driver.call_model(request.clone(), vec![format!("m{i}").into()])), + ); tokio::select! { _ = offloads => Err(test_error("offloads unexpectedly completed")), _ = self.all_started.notified() => { diff --git a/crates/libsy/src/core/processor.rs b/crates/libsy/src/core/processor.rs index d6047e259..af03bdb89 100644 --- a/crates/libsy/src/core/processor.rs +++ b/crates/libsy/src/core/processor.rs @@ -3,7 +3,7 @@ use crate::Result; use async_trait::async_trait; -use switchyard_protocol::{AggLlmResponse, Decision, Request}; +use switchyard_protocol::{AggLlmResponse, ModelId, Request}; /// An event observed by the algorithm. Events are consumed by [`Processor`] to mutate state. /// @@ -20,8 +20,8 @@ pub enum Event<'a> { Decision { /// The request, rewritable in place. request: &'a mut Request, - /// The routing decision produced for `request`. - decision: &'a Decision, + /// The model selected for `request`. + selected_model_id: &'a ModelId, }, /// A buffered response received back from a model. ModelResponse(&'a AggLlmResponse), @@ -82,7 +82,7 @@ mod tests { let mut state = TestState::default(); let mut req = request(); let response = text_response(None, "ok"); - let decision = Decision::new("test/model", true); + let selected_model_id = ModelId::from("test/model"); // Feed one of every event variant through the processor. processor .process(&mut state, Event::Request(&mut req)) @@ -95,7 +95,7 @@ mod tests { &mut state, Event::Decision { request: &mut req, - decision: &decision, + selected_model_id: &selected_model_id, }, ) .await?; diff --git a/crates/libsy/src/core/testing.rs b/crates/libsy/src/core/testing.rs index deee05650..3fdb8c38b 100644 --- a/crates/libsy/src/core/testing.rs +++ b/crates/libsy/src/core/testing.rs @@ -6,7 +6,8 @@ //! libsy offloads every model call, so a test needs something to answer them. [`drive`] is //! [`crate::drive`] with the promise handling filled in: a test supplies a [`Serve`] closure //! standing in for the client a real host would use, and gets back the same -//! `(trace, response)` pair a host does — the same path `switchyard-llm-client`'s `run` +//! `(selected model, response)` pair a host does — the same path +//! `switchyard-llm-client`'s `run` //! takes over HTTP. //! //! The closure is async so a fake can block on a barrier, wait on a notify, or never @@ -16,9 +17,7 @@ use std::future::Future; use std::sync::Arc; use futures::future::BoxFuture; -use switchyard_protocol::{ - Decision, LlmClientError, LlmResponse, ModelId, Request, Response, text_response, -}; +use switchyard_protocol::{LlmClientError, LlmResponse, ModelId, Request, Response, text_response}; use crate::core::algorithm::{Algorithm, CallModel}; use crate::{LibsyError, Result}; @@ -47,12 +46,22 @@ pub(crate) async fn test_drive( algorithm: Arc, request: Request, serve: impl Serve, -) -> Result<(Vec, Response)> { +) -> Result<(ModelId, Response)> { let serve = Arc::new(serve); - crate::drive(algorithm, request, move |call| { - fulfill(Arc::clone(&serve), call) + let routing_serve = Arc::clone(&serve); + let outcome = crate::drive(algorithm, request, move |call| { + fulfill(Arc::clone(&routing_serve), call) }) - .await + .await?; + let selected_model = outcome.selected_model_id.clone(); + let response = match outcome.response { + Some(response) => response, + None => serve + .serve(selected_model.clone(), outcome.request) + .await + .map_err(|source| LibsyError::client_call(selected_model.clone(), source))?, + }; + Ok((selected_model, response)) } /// Serve one call and fulfill its promise, mapping failures the way a host does so diff --git a/crates/libsy/src/error.rs b/crates/libsy/src/error.rs index 7cb47b06d..40eceb829 100644 --- a/crates/libsy/src/error.rs +++ b/crates/libsy/src/error.rs @@ -36,8 +36,8 @@ pub enum LibsyError { #[error(transparent)] Driver(#[from] DriverError), - /// An algorithm's step stream ended without a terminal response. - #[error("algorithm run ended without a final response")] + /// An algorithm's step stream ended without a terminal routing outcome. + #[error("algorithm run ended without a routing outcome")] MissingFinalResponse, /// A target's protocol client failed while serving a routed request. @@ -92,12 +92,6 @@ pub enum DriverError { /// One side of a response promise was dropped before delivery. #[error("driver response promise was dropped")] ResponseDropped, - - /// The consumer took the call's contents with `CallModel::into_parts` instead of - /// answering it, so the run ends here by the consumer's choice. Distinct from - /// [`ResponseDropped`](Self::ResponseDropped), which means the promise was lost. - #[error("model call was abandoned by the consumer")] - Abandoned, } #[cfg(test)] diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 39d93dce3..1bb46db73 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -5,7 +5,7 @@ #![doc = include_str!("../README.md")] mod core; -pub use core::algorithm::{Algorithm, CallModel, Driver, Step, StepStream, drive}; +pub use core::algorithm::{Algorithm, CallModel, Driver, RoutingOutcome, Step, StepStream, drive}; pub use core::classifier::{Classification, Classifier, Score}; pub use core::processor::{Event, Processor}; pub use core::state::{State, StateValue}; @@ -41,12 +41,3 @@ pub use algorithms::util::stage::{ }; mod observability; - -/// Registers process-wide compatibility gauges with the global meter provider. -/// -/// Hosts should call this after installing their OpenTelemetry meter provider. Other -/// libsy metrics are created when recorded and do not require initialization. Spans -/// and structured logs require the host to install a `tracing` subscriber separately. -pub fn initialize_metrics() { - observability::initialize_metrics(); -} diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 7eb7f3891..f43f02bfb 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -5,7 +5,7 @@ //! algorithm layer. //! //! [`Algorithm::run_stream`](crate::Algorithm::run_stream) and [`Driver`] call these -//! helpers around the [`Decision`] hook and the offload boundary, so every algorithm is +//! helpers around the routing outcome and the offload boundary, so every algorithm is //! instrumented from the outside and carries no telemetry code of its own. The provider //! call on the other side of the offload belongs to the host, and is instrumented by //! whoever makes it. Metrics record through the @@ -33,59 +33,27 @@ //! negligible next to a model call. use std::future::Future; -use std::sync::OnceLock; -use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; -use opentelemetry::metrics::{Meter, ObservableGauge}; +use opentelemetry::metrics::Meter; use opentelemetry::{KeyValue, global}; use tracing::Span; -use crate::{DriverError, LibsyError, Result}; -use switchyard_protocol::{Decision, Request, Response}; +use crate::Result; +use switchyard_protocol::{ModelId, Request, Response}; const METRICS_SCOPE: &str = "switchyard"; const TRACING_TARGET: &str = "libsy"; -static TOTAL_REQUESTS: AtomicU64 = AtomicU64::new(0); -static TOTAL_ERRORS: AtomicU64 = AtomicU64::new(0); -static TOTAL_GAUGES: OnceLock<(ObservableGauge, ObservableGauge)> = OnceLock::new(); - /// The `libsy`-scoped meter from the globally installed provider. pub(crate) fn meter() -> Meter { global::meter(METRICS_SCOPE) } -/// Registers process-wide compatibility gauges with the installed global meter provider. -pub(crate) fn initialize_metrics() { - TOTAL_GAUGES.get_or_init(|| { - let meter = meter(); - let requests = meter - .u64_observable_gauge("switchyard.total_requests") - .with_callback(|observer| { - observer.observe(TOTAL_REQUESTS.load(Ordering::Relaxed), &[]); - }) - .build(); - let errors = meter - .u64_observable_gauge("switchyard.total_errors") - .with_callback(|observer| { - observer.observe(TOTAL_ERRORS.load(Ordering::Relaxed), &[]); - }) - .build(); - (requests, errors) - }); -} - -/// Whether a result is a call the consumer deliberately abandoned rather than a failure. -pub(crate) fn is_abandoned(result: &Result) -> bool { - matches!(result, Err(LibsyError::Driver(DriverError::Abandoned))) -} - -/// `outcome` attribute value for a result: `ok`, `abandoned`, or `error`. +/// `outcome` attribute value for a result: `ok` or `error`. pub(crate) fn outcome_value(result: &Result) -> &'static str { match result { Ok(_) => "ok", - Err(LibsyError::Driver(DriverError::Abandoned)) => "abandoned", Err(_) => "error", } } @@ -145,10 +113,10 @@ pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span { /// Runs one algorithm task to completion, recording the run counter, duration /// histogram, span outcome, and failure log when it resolves. /// Executes inside the `libsy.run` span its caller instruments the task with. -pub(crate) async fn observe_run( +pub(crate) async fn observe_run( algorithm: &str, - run: impl Future>, -) -> Result { + run: impl Future>, +) -> Result { let started = Instant::now(); let result = run.await; let duration = started.elapsed(); @@ -159,14 +127,10 @@ pub(crate) async fn observe_run( /// Records the end of one algorithm run: the run counter and duration /// histogram, the `outcome`/`error` fields on `span`, and a warn log when the /// run failed. -fn record_run(algorithm: &str, duration: Duration, result: &Result, span: &Span) { +fn record_run(algorithm: &str, duration: Duration, result: &Result, span: &Span) { let outcome = outcome_value(result); span.record("outcome", outcome); - // An abandoned run ended because the consumer took the call, not because anything went - // wrong, so it is counted but not reported as a failure. - if let Err(error) = result - && !is_abandoned(result) - { + if let Err(error) = result { span.record("error", tracing::field::display(error)); tracing::warn!( target: TRACING_TARGET, @@ -211,7 +175,6 @@ pub(crate) fn record_classifier_fail_open(judge_model: &str, reason: &'static st pub(crate) fn record_llm_call( algorithm: &str, selected_model: &str, - is_answer_call: bool, duration: Duration, result: &Result, span: &Span, @@ -234,29 +197,6 @@ pub(crate) fn record_llm_call( .build() .record(duration.as_secs_f64() * 1000.0, &call_attributes); - // An abandoned answer call was never served, so it counts as neither a served - // request nor an error. - if is_answer_call && !is_abandoned(result) { - TOTAL_REQUESTS.fetch_add(1, Ordering::Relaxed); - let routed_attributes = [KeyValue::new("model", selected_model.to_string())]; - if result.is_ok() { - meter - .u64_counter("switchyard.requests") - .build() - .add(1, &routed_attributes); - meter - .f64_histogram("switchyard.model_call_latency_ms") - .build() - .record(duration.as_secs_f64() * 1000.0, &routed_attributes); - } else { - TOTAL_ERRORS.fetch_add(1, Ordering::Relaxed); - meter - .u64_counter("switchyard.errors") - .build() - .add(1, &routed_attributes); - } - } - match result { Ok(response) => { // Token usage exists only once a response is buffered; a streamed @@ -275,7 +215,7 @@ pub(crate) fn record_llm_call( } } } - Err(error) if !is_abandoned(result) => { + Err(error) => { span.record("error", tracing::field::display(error)); tracing::warn!( target: TRACING_TARGET, @@ -285,13 +225,11 @@ pub(crate) fn record_llm_call( "model call failed" ); } - Err(_) => {} } } /// Records one published routing decision: the decision counter plus a structured debug event. -pub(crate) fn record_decision(algorithm: &str, decision: &Decision) { - let selected_model = decision.selected_model_id(); +pub(crate) fn record_decision(algorithm: &str, selected_model: &ModelId) { tracing::debug!( target: TRACING_TARGET, algorithm, diff --git a/crates/protocol/README.md b/crates/protocol/README.md index f3dfd891b..7e47dcf02 100644 --- a/crates/protocol/README.md +++ b/crates/protocol/README.md @@ -21,7 +21,7 @@ serde_json = "1" | Response | [`AggLlmResponse`], [`ResponseOutput`], [`Usage`], [`StopReason`] | | Streaming | [`LlmResponse`], [`LlmResponseStream`], [`LlmResponseStreamEvent`], [`LlmResponseChunk`], [`ProviderStreamEvent`] | | Envelope | [`Request`], [`Response`], [`Metadata`] | -| Routing I/O | [`Decision`], [`RoutedLlmClient`], [`LlmClientError`] | +| Routing I/O | [`RoutedLlmClient`], [`LlmClientError`] | | Wire identity | [`WireFormat`], [`FormatId`] | ## Simple request diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index dd9f29236..6b24afcc6 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -1,14 +1,12 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! The routed-call server trait and the routing decision it carries. +//! The routed-call server trait and its shared error types. //! //! [`RoutedLlmClient`] is the one piece of I/O the protocol does not own: a host -//! implements it to actually perform a model call. [`Decision`] is the routing -//! decision that produced the call, carried alongside so the client and any -//! observer can see which model was chosen and why. Both live here — rather than -//! in libsy's orchestration crate — so a client crate that depends only on the -//! protocol can serve routed calls without pulling in the orchestrator. +//! implements it to actually perform a model call. It lives here — rather than in +//! libsy's orchestration crate — so a client crate that depends only on the protocol +//! can serve routed calls without pulling in the orchestrator. use async_trait::async_trait; use thiserror::Error; @@ -127,35 +125,6 @@ impl RoutingFallbackReason { } } -/// A routing choice produced by an algorithm. -#[derive(Clone, Debug)] -pub struct Decision { - /// The model identifier selected for the call. - selected_model_id: ModelId, - /// True for an answer-generating call. False for classifier and judge calls. - is_answer_call: bool, -} - -impl Decision { - /// Creates a decision and records whether its call produces the answer. - pub fn new(selected_model_id: impl Into, is_answer_call: bool) -> Self { - Self { - selected_model_id: selected_model_id.into(), - is_answer_call, - } - } - - /// The model identifier selected for the call. - pub fn selected_model_id(&self) -> &ModelId { - &self.selected_model_id - } - - /// Whether this call generates an answer rather than a routing verdict. - pub fn is_answer_call(&self) -> bool { - self.is_answer_call - } -} - /// Performs the actual model call for a target. This is the one piece of I/O the /// library does not own — a host implements it over its own transport (HTTP SDK, /// in-process model, mock). It serves a call the stream consumer chose not to diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 5d6e018c1..4546c549d 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -15,11 +15,11 @@ use switchyard_libsy::{ Algorithm, CallModel, ClassifierContractConfig, ClassifierResponseFormat, CustomClassifierConfig, CustomClassifierPolicy, EscalationJudgeConfig, HandoffNoteConfig, LibsyError as RustLibsyError, LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, - PickerMode, Random, StageRouter, StageRouterConfig, Step as RustStep, StepStream, - TaskClassifierConfig, + PickerMode, Random, RoutingOutcome, StageRouter, StageRouterConfig, Step as RustStep, + StepStream, TaskClassifierConfig, }; use switchyard_protocol::{ - AggLlmResponse, Decision, LlmClientError, LlmResponse, Metadata, ModelId, Request, Response, + AggLlmResponse, LlmClientError, LlmResponse, Metadata, ModelId, Request, Response, }; use tokio::sync::Mutex; @@ -335,41 +335,6 @@ impl PyLlmFallback { } } -/// A routing choice produced by an algorithm. -#[pyclass(name = "Decision", module = "switchyard.libsy", frozen)] -struct PyDecision { - inner: Decision, -} - -impl From for PyDecision { - fn from(inner: Decision) -> Self { - Self { inner } - } -} - -#[pymethods] -impl PyDecision { - /// The semantic model id selected for the call. - #[getter] - fn selected_model_id(&self) -> &str { - self.inner.selected_model_id().as_str() - } - - /// Whether this call produces the answer rather than a routing verdict. - #[getter] - fn is_answer_call(&self) -> bool { - self.inner.is_answer_call() - } - - fn __repr__(&self) -> String { - format!( - "Decision(selected_model_id={:?}, is_answer_call={})", - self.inner.selected_model_id(), - self.inner.is_answer_call() - ) - } -} - /// One model call yielded by [`PyAlgorithm::run_stream`]. #[pyclass(name = "ModelCall", module = "switchyard.libsy")] struct PyModelCall { @@ -377,28 +342,16 @@ struct PyModelCall { algorithm: String, request: Py, models: Vec, - decision: Py, } impl PyModelCall { fn new(py: Python<'_>, call: CallModel) -> PyResult { let request = to_python(py, &call.request.llm_request)?; - let selected = call - .models - .first() - .cloned() - .ok_or(RustLibsyError::NoTargets) - .map_err(py_libsy_error)?; - let decision = Py::new( - py, - PyDecision::from(Decision::new(selected, call.is_answer_call)), - )?; Ok(Self { algorithm: call.algorithm.clone(), models: call.models.iter().map(ToString::to_string).collect(), inner: Some(call), request, - decision, }) } @@ -429,20 +382,6 @@ impl PyModelCall { self.models.clone() } - /// The routing decision behind this call. - #[getter] - fn decision(&self, py: Python<'_>) -> Py { - self.decision.clone_ref(py) - } - - /// Consume the answer call without serving it and return its rewritten request and decision. - #[pyo3(name = "into_parts")] - fn take_parts(&mut self, py: Python<'_>) -> PyResult<(Py, Py)> { - let decision = self.decision.clone_ref(py); - let (request, _models) = self.take()?.into_parts(); - Ok((to_python(py, &request.llm_request)?, decision)) - } - /// Fulfill this call with an aggregate normalized response dictionary. fn respond(&mut self, response: &Bound<'_, PyAny>) -> PyResult<()> { let aggregate = from_python::(response)?; @@ -482,15 +421,51 @@ impl PyModelCall { } } +/// The terminal routing selection, rewritten request, and optional existing response. +#[pyclass(name = "RoutingOutcome", module = "switchyard.libsy", frozen)] +struct PyRoutingOutcome { + selected_model_id: String, + fallback_models: Vec, + request: Py, + response: Option>, +} + +#[pymethods] +impl PyRoutingOutcome { + /// The model selected by the algorithm and tried first by the host. + #[getter] + fn selected_model_id(&self) -> &str { + &self.selected_model_id + } + + /// Additional models the host may try in order after an eligible failure. + #[getter] + fn fallback_models(&self) -> Vec { + self.fallback_models.clone() + } + + /// The normalized request after routing-time rewrites. + #[getter] + fn request(&self, py: Python<'_>) -> Py { + self.request.clone_ref(py) + } + + /// An answer produced while routing, when one already exists. + #[getter] + fn response(&self, py: Python<'_>) -> Option> { + self.response + .as_ref() + .map(|response| response.clone_ref(py)) + } +} + /// One item yielded by a Python algorithm stream. #[pyclass(name = "Step", module = "switchyard.libsy", frozen)] enum PyStep { /// The host must serve the model call before the algorithm can continue. CallModel { call: Py }, - /// A routing decision emitted by the algorithm. - Decision { decision: Py }, - /// The terminal aggregate response. - Done { response: Py }, + /// The terminal routing outcome. + Done { outcome: Py }, } /// Async Python iterator over one Rust algorithm run. @@ -526,7 +501,7 @@ struct PyAlgorithm { #[pymethods] impl PyAlgorithm { - /// Run the algorithm as a stream of model calls, decisions, and one terminal response. + /// Run the algorithm as routing-time model calls followed by one terminal outcome. /// /// `headers`, when given, is normalized into the request's correlation /// [`Metadata`] exactly as an HTTP host would (`Metadata::from_headers`), @@ -565,20 +540,40 @@ async fn step_to_python(step: RustStep) -> PyResult { call: Py::new(py, PyModelCall::new(py, *call)?)?, }) }), - RustStep::Decision(decision) => Python::attach(|py| { - Ok(PyStep::Decision { - decision: Py::new(py, PyDecision::from(decision))?, - }) - }), - RustStep::Done(response) => { - let response = response - .llm_response - .into_agg() - .await - .map_err(py_libsy_error)?; + RustStep::Done(outcome) => { + let RoutingOutcome { + selected_model_id, + fallback_models, + request, + response, + } = *outcome; + let response = match response { + Some(response) => Some( + response + .llm_response + .into_agg() + .await + .map_err(py_libsy_error)?, + ), + None => None, + }; Python::attach(|py| { Ok(PyStep::Done { - response: to_python(py, &response)?, + outcome: Py::new( + py, + PyRoutingOutcome { + selected_model_id: selected_model_id.to_string(), + fallback_models: fallback_models + .iter() + .map(ToString::to_string) + .collect(), + request: to_python(py, &request.llm_request)?, + response: response + .as_ref() + .map(|response| to_python(py, response)) + .transpose()?, + }, + )?, }) }) } @@ -730,12 +725,12 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { let libsy_module = PyModule::new(module.py(), "libsy")?; libsy_module.add_class::()?; libsy_module.add_class::()?; - libsy_module.add_class::()?; libsy_module.add_class::()?; libsy_module.add_class::()?; libsy_module.add_class::()?; libsy_module.add_class::()?; libsy_module.add_class::()?; + libsy_module.add_class::()?; libsy_module.add_class::()?; libsy_module.add_class::()?; libsy_module.add_function(wrap_pyfunction!(noop_algorithm, &libsy_module)?)?; diff --git a/crates/switchyard-server/README.md b/crates/switchyard-server/README.md index b853929d7..cf107fecc 100644 --- a/crates/switchyard-server/README.md +++ b/crates/switchyard-server/README.md @@ -159,13 +159,15 @@ Routed-call compatibility metrics are: | `switchyard_requests_total` | counter | `model` | Successful final routed calls | | `switchyard_errors_total` | counter | `model` | Failed final routed calls | | `switchyard_model_call_latency_ms` | histogram | `model` | Successful final routed-call latency | +| `switchyard_llm_calls_total` | counter | `algorithm`, `selected_model`, `outcome` | Logical routing and terminal model calls | +| `switchyard_llm_call_duration_ms` | histogram | `algorithm`, `selected_model`, `outcome` | Logical routing and terminal model-call latency | | `switchyard_prompt_tokens_total` | counter | `model` | Input tokens, including cached and cache-creation tokens | | `switchyard_completion_tokens_total` | counter | `model` | Output tokens | | `switchyard_cached_tokens_total` | counter | `model` | Cached input tokens | | `switchyard_cache_creation_tokens_total` | counter | `model` | Cache-creation input tokens | | `switchyard_reasoning_tokens_total` | counter | `model` | Reasoning output tokens | | `switchyard_total_latency_ms` | histogram | `model` | Full-turn latency for successful routed responses | -| `switchyard_routing_overhead_ms` | histogram | `algorithm` | Algorithm run time minus the call that served it | +| `switchyard_routing_overhead_ms` | histogram | `algorithm` | Time spent producing the terminal routing outcome | | `switchyard_classifier_fail_open_total` | counter | `judge_model`, `reason` | Judge failures that made a classifier route without a verdict | | `switchyard_client_responses_total` | counter | `outcome` | Final LLM-route responses | | `switchyard_upstream_attempts_total` | counter | `outcome`, `code` | Actual upstream HTTP attempts | @@ -175,21 +177,24 @@ Routed-call compatibility metrics are: judge call failed. `judge_model` names the configured judge target, and `reason` is one of eight fixed error categories. +`switchyard_llm_calls_total` and `switchyard_llm_call_duration_ms` retain the logical call +boundary across routing: libsy records classifier and judge calls, while `libsy-llm-client` +records the terminal call when the routing outcome requires one. A response already produced by +routing is counted only by its original libsy call. Terminal fallback and backend retries remain +one logical call labeled with the algorithm-selected model. + `switchyard_total_latency_ms` observes an aggregate when it becomes available or a stream when it ends cleanly. Its clock starts in a router-wide middleware, before the request body is read and decoded, so it measures request ingress through response completion. It still excludes connection accept and TLS handshake, which hyper completes before the server sees the request. -`switchyard_routing_overhead_ms` is what routing cost on top of the model call: the algorithm's run -time minus the call that served the request. Classifier calls are not subtracted, so an -LLM-classifier route reports its classification time here while `passthrough` and `random` report -the sub-millisecond cost of picking a target. It carries only `algorithm`, since the number -describes the router and not the target it chose, and a run that served nothing records nothing. Its -buckets start at 0.1 ms via a view in the server; the SDK defaults start at 5 ms. - -Both clocks stop when the routed call resolves, which for a streamed response is when the stream -handle arrives rather than when the stream ends, so SSE relay time is in neither term. The Python -summary of the same name measures its total through stream completion, making its streaming values -mostly generation time. +`switchyard_routing_overhead_ms` records the elapsed time in `run` from `run_started` until +`drive` returns a routing outcome. It includes algorithm execution and any classifier or judge +calls made while driving the algorithm, and it is recorded before a terminal answer call begins. +No duration for the request-serving call is subtracted. If routing itself produced the answer, +that call occurred inside the measured `drive` interval. The metric carries only `algorithm`, +since the duration describes the router rather than the target it chose; a routing failure before +an outcome records nothing. Its buckets start at 0.1 ms via a view in the server; the SDK defaults +start at 5 ms. See [CONFIGURATION.md](CONFIGURATION.md) to add an LLM client, target, or algorithm. diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 4e980152f..24f1026cc 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -451,15 +451,17 @@ fn stats_observer( classifier_log: Option<(SharedRoutingLog, routing_log::RoutingLogContext)>, ) -> RunObserver { Arc::new(move |observation| match observation { + RunObservation::AnswerCall(call) => { + let latency_ms = call.duration.as_secs_f64() * 1_000.0; + if call.is_success { + stats.record_success(&call.selected_model, latency_ms); + } else { + stats.record_error(&call.selected_model); + } + } RunObservation::LlmCall(call) => { let latency_ms = call.duration.as_secs_f64() * 1_000.0; - if call.is_answer_call { - if call.is_success { - stats.record_success(&call.selected_model, latency_ms); - } else { - stats.record_error(&call.selected_model); - } - } else if call.is_success { + if call.is_success { if let (Some((log, context)), Some(usage)) = (classifier_log.as_ref(), call.usage.as_ref()) { @@ -752,18 +754,14 @@ async fn handle_llm_request( state.stats.clone(), state.routing_log.clone().zip(routing_log_context.clone()), ); - let (trace, response) = + let (selected_model, response) = match switchyard_llm_client::run(algorithm, client_router, request, Some(observer)).await { Ok(result) => result, Err(error) => return algorithm_error(error), }; - let decision = trace.last(); // The response carries the candidate that actually served it. Fall back to the routing - // decision for algorithms that return a response without an offloaded model call. - let served_model = response - .served_model() - .cloned() - .or_else(|| decision.map(|decision| decision.selected_model_id().clone())); + // selection for algorithms that return a response without an offloaded model call. + let served_model = response.served_model().cloned().or(Some(selected_model)); let response = if let Some(served_model) = served_model.as_ref() { let cache_eligible = cache_probe .as_ref() @@ -1373,10 +1371,9 @@ mod tests { let context = routing_log::RoutingLogContext::from_metadata(&metadata); let observer = stats_observer(StatsAccumulator::default(), Some((log.clone(), context))); - let call = |model: &str, is_answer_call: bool| { - RunObservation::LlmCall(LlmCallObservation { + let call = |model: &str, answer: bool| { + let observation = LlmCallObservation { selected_model: ModelId::from(model), - is_answer_call, is_success: true, duration: Duration::from_millis(3), usage: Some(Usage { @@ -1384,7 +1381,12 @@ mod tests { output_tokens: Some(7), ..Usage::default() }), - }) + }; + if answer { + RunObservation::AnswerCall(observation) + } else { + RunObservation::LlmCall(observation) + } }; observer(call("judge-model", false)); observer(call("routed-model", true)); diff --git a/crates/switchyard-server/src/metrics.rs b/crates/switchyard-server/src/metrics.rs index b6ba04f5c..41bd50f5c 100644 --- a/crates/switchyard-server/src/metrics.rs +++ b/crates/switchyard-server/src/metrics.rs @@ -61,7 +61,7 @@ fn initialize() -> Result { } let provider = builder.build(); global::set_meter_provider(provider.clone()); - libsy::initialize_metrics(); + switchyard_llm_client::initialize_metrics(); global::meter("switchyard") .u64_gauge("switchyard.build_info") .build() diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index cff041ae0..3992837bf 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -510,7 +510,7 @@ async fn stats_accumulates_buffered_success_error_and_shared_routes() -> TestRes assert_eq!(stats["models"]["gemini-3.5-flash"]["calls"], 1); assert_eq!(stats["models"]["gemini-3.5-flash"]["errors"], 1); assert_eq!(stats["models"]["model/unknown"]["calls"], 1); - assert_eq!(stats["routing_overhead"]["count"], 2); + assert_eq!(stats["routing_overhead"]["count"], 3); Ok(()) } @@ -606,9 +606,7 @@ async fn metrics_exposes_switchyard_otel_instruments() -> TestResult { "switchyard_client_responses_total{outcome=\"success\",", "switchyard_upstream_attempts_total{code=\"200\",outcome=\"success\",", "# TYPE switchyard_runs_total counter", - "# TYPE switchyard_llm_calls_total counter", "# TYPE switchyard_run_duration_ms histogram", - "# TYPE switchyard_llm_call_duration_ms histogram", "# TYPE switchyard_prompt_tokens_total counter", "# TYPE switchyard_completion_tokens_total counter", "# TYPE switchyard_cached_tokens_total counter", @@ -1313,6 +1311,83 @@ prompt = "CUSTOM STAGE" Ok(()) } +#[tokio::test] +async fn accepted_escalation_response_is_logged_once_as_the_final_answer() -> TestResult { + let upstream = MockUpstream::start().await?; + let temp_dir = tempfile::tempdir()?; + let state = load_test_config(&format!( + r#" +schema_version = 1 + +[llm_clients.upstream] +format = "openai_chat" +base_url = "{base_url}" + +[targets.classifier] +id = "model/classifier" +llm_client = "upstream" + +[targets.strong] +id = "model/strong" +llm_client = "upstream" + +[targets.weak] +id = "model/weak" +llm_client = "upstream" + +[routes.escalation] +id = "switchyard/escalation" +type = "llm_classifier" +mode = "escalation" +classifier_target = "classifier" +strong_target = "strong" +weak_target = "weak" +escalation = {{ confirmations = 1 }} +"#, + base_url = upstream.base_url + ))? + .with_routing_log(temp_dir.path().join("routing.jsonl"))?; + let app = build_switchyard_router(state); + + let response = send_with_headers( + &app, + "POST", + "/v1/chat/completions", + Some(json!({ + "model": "switchyard/escalation", + "messages": [{"role": "user", "content": "bounded task"}] + })), + &[("x-switchyard-session-id", "accepted-escalation")], + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(upstream.models().await, ["model/weak", "model/classifier"]); + + let stats = send( + &app, + "GET", + "/v1/routing/session-stats?session_id=accepted-escalation", + None, + ) + .await?; + assert_eq!(stats.status, StatusCode::OK); + let stats = stats.json()?; + assert_eq!(stats["total_calls"], 2); + assert_eq!(stats["total_prompt_tokens"], 20); + assert_eq!(stats["total_completion_tokens"], 4); + assert_eq!(stats["models"]["model/weak"]["calls"], 1); + assert_eq!(stats["models"]["model/classifier"]["calls"], 1); + + let process_stats = send(&app, "GET", "/v1/stats", None).await?.json()?; + assert_eq!(process_stats["total_requests"], 1); + assert_eq!(process_stats["models"]["model/weak"]["calls"], 1); + assert_eq!( + process_stats["models"]["model/weak"]["model_call_latency"]["count"], + 1 + ); + Ok(()) +} + #[tokio::test] async fn stage_classifier_can_request_json_object_output() -> TestResult { let upstream = MockUpstream::start().await?; @@ -2850,8 +2925,9 @@ async fn advisor_route_redo_fail_open_and_stats_projection() -> TestResult { ); let stats = send(&app, "GET", "/v1/stats", None).await?.json()?; - // State-owned accumulator: three executor answer calls, one failed consult. - assert_eq!(stats["models"]["model/executor"]["calls"], 3); + // State-owned accumulator: two client-visible executor answers; the discarded REDO attempt + // is routing work. The failed advisor consult is counted separately. + assert_eq!(stats["models"]["model/executor"]["calls"], 2); assert_eq!(stats["classifier"]["total_errors"], 1); // Projection deltas for the metrics only this test emits. let redo = gate_count(&stats, &["reviews", "redo", "total"]) diff --git a/docs/getting_started.md b/docs/getting_started.md index c6428e970..626a58adc 100644 --- a/docs/getting_started.md +++ b/docs/getting_started.md @@ -268,11 +268,12 @@ behaviour. ### Drive the algorithm -An algorithm yields a stream of steps. Each `Step::CallModel` is a model call your -host performs over its own transport, and the run ends with -`Step::Done` carrying the final response. Serving those calls yourself -is what lets libsy embed in a host that already owns its HTTP stack, retries, -and credentials. +An algorithm yields a stream of steps. Each `Step::CallModel` is a routing-time classifier or +judge call your host performs over its own transport. The run ends with `Step::Done` carrying a +`RoutingOutcome`: the selected model, ordered fallbacks, rewritten request, and an optional +response when routing already produced the answer. Otherwise the host makes the terminal answer +call from that outcome. Serving these calls yourself is what lets libsy embed in a host that +already owns its HTTP stack, retries, and credentials. For the request, response, and streaming types the steps carry, see [`switchyard-protocol`](../crates/protocol/README.md). diff --git a/switchyard/libsy/__init__.py b/switchyard/libsy/__init__.py index 69e5f26c5..d6c2d243b 100644 --- a/switchyard/libsy/__init__.py +++ b/switchyard/libsy/__init__.py @@ -7,12 +7,12 @@ Algorithm, ContextWindowExceededError, CustomClassifierConfig, - Decision, EscalationClassifierConfig, LibsyError, LlmClassifierConfig, LlmFallback, ModelCall, + RoutingOutcome, Step, TaskClassifierConfig, ) @@ -23,12 +23,12 @@ "Algorithm", "ContextWindowExceededError", "CustomClassifierConfig", - "Decision", "EscalationClassifierConfig", "LibsyError", "LlmClassifierConfig", "LlmFallback", "ModelCall", + "RoutingOutcome", "Step", "TaskClassifierConfig", "algorithms", diff --git a/switchyard_rust/libsy.py b/switchyard_rust/libsy.py index 65542dd68..c7037717b 100644 --- a/switchyard_rust/libsy.py +++ b/switchyard_rust/libsy.py @@ -15,12 +15,12 @@ "Algorithm", "ContextWindowExceededError", "CustomClassifierConfig", - "Decision", "EscalationClassifierConfig", "LibsyError", "LlmClassifierConfig", "LlmFallback", "ModelCall", + "RoutingOutcome", "Step", "TaskClassifierConfig", "llm_classifier", @@ -79,39 +79,33 @@ def __init__( ) -> None: ... @final - class Decision: - """A semantic routing choice produced by an algorithm.""" - + class ModelCall: @property - def selected_model_id(self) -> str: ... + def algorithm(self) -> str: ... @property - def reasoning(self) -> str | None: ... + def request(self) -> dict[str, object]: ... @property - def is_answer_call(self) -> bool: ... + def models(self) -> list[str]: ... - _RoutingDecision = Decision + def respond(self, response: Mapping[str, object]) -> None: ... + + def fail(self, error: BaseException) -> None: ... @final - class ModelCall: + class RoutingOutcome: @property - def algorithm(self) -> str: ... + def selected_model_id(self) -> str: ... @property - def request(self) -> dict[str, object]: ... + def fallback_models(self) -> list[str]: ... @property - def models(self) -> list[str]: ... + def request(self) -> dict[str, object]: ... @property - def decision(self) -> Decision: ... - - def into_parts(self) -> tuple[dict[str, object], Decision]: ... - - def respond(self, response: Mapping[str, object]) -> None: ... - - def fail(self, error: BaseException) -> None: ... + def response(self) -> dict[str, object] | None: ... class Step: @final @@ -119,15 +113,10 @@ class CallModel: __match_args__: ClassVar[tuple[Literal["call"]]] = ("call",) call: ModelCall - @final - class Decision: - __match_args__: ClassVar[tuple[Literal["decision"]]] = ("decision",) - decision: _RoutingDecision - @final class Done: - __match_args__: ClassVar[tuple[Literal["response"]]] = ("response",) - response: dict[str, object] + __match_args__: ClassVar[tuple[Literal["outcome"]]] = ("outcome",) + outcome: RoutingOutcome @final class TaskClassifierConfig: @@ -205,7 +194,7 @@ def run_stream( self, request: Mapping[str, object], headers: Mapping[str, str] | None = None, - ) -> AsyncIterator[Step.CallModel | Step.Decision | Step.Done]: ... + ) -> AsyncIterator[Step.CallModel | Step.Done]: ... def noop() -> Algorithm: ... diff --git a/tests/test_libsy_minimal_bindings.py b/tests/test_libsy_minimal_bindings.py index a01d61d03..ae2a9c7b4 100644 --- a/tests/test_libsy_minimal_bindings.py +++ b/tests/test_libsy_minimal_bindings.py @@ -11,9 +11,8 @@ Algorithm, ContextWindowExceededError, CustomClassifierConfig, - Decision, - LibsyError, LlmClassifierConfig, + RoutingOutcome, Step, TaskClassifierConfig, algorithms, @@ -57,8 +56,7 @@ async def run_algorithm( *, request: dict[str, Any] | None = None, headers: dict[str, str] | None = None, -) -> tuple[list[Decision], dict[str, Any]]: - decisions: list[Decision] = [] +) -> tuple[str, dict[str, Any]]: async for step in algorithm.run_stream(request or request_body(), headers=headers): match step: case Step.CallModel(call): @@ -76,66 +74,49 @@ async def run_algorithm( else: call.respond(response) break - case Step.Decision(decision): - decisions.append(decision) - case Step.Done(response): - return decisions, response - raise AssertionError("algorithm stream ended without a response") + case Step.Done(outcome): + if outcome.response is not None: + return outcome.selected_model_id, outcome.response + candidates = [outcome.selected_model_id, *outcome.fallback_models] + for index, target in enumerate(candidates): + candidate_request = {**outcome.request, "model": target} + client = (clients or {})[target] + try: + response = await client.call(candidate_request) + except ContextWindowExceededError: + if index + 1 == len(candidates): + raise + else: + return outcome.selected_model_id, response + raise AssertionError("algorithm stream ended without an outcome") async def test_random_streams_complex_steps_and_accepts_a_dictionary_response() -> None: client = EchoClient("fast") algorithm = algorithms.random(["fast"]) - decisions: list[Decision] = [] - response: dict[str, Any] | None = None + outcome: RoutingOutcome | None = None variants: list[str] = [] async for step in algorithm.run_stream(request_body()): match step: - case Step.CallModel(call): - variants.append("call_model") - assert call.models == ["fast"] - client_response = await client.call(call.request) - call.respond(client_response) - with pytest.raises(LibsyError, match="already been completed"): - call.respond(client_response) - case Step.Decision(decision): - variants.append("decision") - decisions.append(decision) case Step.Done(done): variants.append("done") - response = done - - assert variants == ["decision", "call_model", "done"] - assert len(decisions) == 1 - assert decisions[0].selected_model_id == "fast" - assert decisions[0].is_answer_call is True + outcome = done + + assert variants == ["done"] + assert outcome is not None + assert outcome.selected_model_id == "fast" + assert outcome.fallback_models == [] + assert outcome.response is None + response = await client.call(outcome.request) assert client.calls[0]["model"] == "fast" assert client.calls[0]["messages"][0]["content"] == [ {"type": "text", "text": "hello"} ] - assert response is not None assert response["model"] == "fast" assert response["outputs"][0]["content"] == [{"type": "text", "text": "fast"}] -async def test_into_parts_supports_decision_only_routing() -> None: - algorithm = algorithms.random(["fast"]) - - async for step in algorithm.run_stream(request_body()): - match step: - case Step.CallModel(call) if call.decision.is_answer_call: - request, decision = call.into_parts() - assert call.algorithm == "random" - assert call.models == ["fast"] - assert request["messages"] == request_body()["messages"] - assert decision.selected_model_id == "fast" - assert decision.is_answer_call is True - with pytest.raises(LibsyError, match="already been completed"): - call.into_parts() - break - - async def test_classifier_config_accepts_a_prompt_override() -> None: """Verify that a configured classifier prompt is rendered for the judge.""" @@ -322,9 +303,9 @@ def test_random_rejects_invalid_weights() -> None: async def test_noop_needs_no_client() -> None: - decisions, response = await run_algorithm(algorithms.noop()) + selected_model, response = await run_algorithm(algorithms.noop()) - assert decisions[0].selected_model_id == "auto" + assert selected_model == "auto" assert response["outputs"][0]["content"] == [{"type": "text", "text": "OK"}] @@ -341,11 +322,11 @@ def test_algorithm_rejects_invalid_headers(headers: dict[str, str], message: str async def test_algorithm_accepts_case_insensitive_duplicate_names() -> None: - decisions, _ = await run_algorithm( + selected_model, _ = await run_algorithm( algorithms.noop(), headers={"X-Unused": "first", "x-unused": "second"} ) - assert decisions[0].selected_model_id == "auto" + assert selected_model == "auto" def test_algorithm_rejects_header_map_capacity_overflow() -> None: @@ -379,17 +360,6 @@ def test_invalid_request_is_rejected_at_the_boundary() -> None: ) -async def test_client_failure_becomes_libsy_error() -> None: - class FailingClient: - async def call(self, request: dict[str, Any]) -> dict[str, Any]: - raise RuntimeError("client failed") - - algorithm = algorithms.random(["broken"]) - - with pytest.raises(LibsyError, match="client failed"): - await run_algorithm(algorithm, {"broken": FailingClient()}) - - async def test_context_window_failure_falls_back_to_the_next_model() -> None: class OverflowClient: async def call(self, request: dict[str, Any]) -> dict[str, Any]: @@ -401,10 +371,10 @@ async def call(self, request: dict[str, Any]) -> dict[str, Any]: picker="efficient_first", confidence_threshold=0.5, ) - decisions, response = await run_algorithm( + selected_model, response = await run_algorithm( algorithm, {"fast": OverflowClient(), "strong": EchoClient("strong")}, ) - assert [decision.selected_model_id for decision in decisions] == ["fast"] + assert selected_model == "fast" assert response["model"] == "strong"