diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 8928808e..fc3b4b3b 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -13,7 +13,7 @@ use reqwest::RequestBuilder; use reqwest::header::{HeaderMap, RETRY_AFTER}; use serde_json::{Map, Value}; use switchyard_protocol::{ - Decision, LlmRequest, LlmResponse, Metadata, ModelId, Request, Response, RoutedLlmClient, + LlmRequest, LlmResponse, Metadata, ModelId, Request, Response, RoutedLlmClient, }; use switchyard_translation::{ WireFormat, decode_aggregated_response, decode_request, decode_stream, @@ -144,10 +144,11 @@ impl TranslatingLlmClient { message: format!("model {model} has no Anthropic backend for count_tokens"), })?; let Request { - llm_request, + mut llm_request, metadata, .. } = request; + llm_request.model = Some(model.to_string()); let http_response = self .send_encoded( backend, @@ -171,9 +172,9 @@ impl TranslatingLlmClient { }) } - /// Encode `llm_request` (its model restamped to `model`) for `wire_format`, - /// POST it to `url` with the request's forwarded headers plus the backend's - /// static headers and auth, and return the successful upstream response. A + /// Encode `llm_request` for `wire_format`, POST it to `url` with the request's + /// forwarded headers plus the backend's static headers and auth, and return the + /// successful upstream response. A /// buffered response is fully collected within the retry boundary; a streamed /// response is returned as soon as its successful headers arrive. A non-success /// status maps to a typed error — a 400 is classified as a context-window @@ -186,13 +187,11 @@ impl TranslatingLlmClient { &self, backend: &Backend, wire_format: WireFormat, - mut llm_request: LlmRequest, + llm_request: LlmRequest, metadata: Option<&Metadata>, model: &ModelId, endpoint: UpstreamEndpoint, ) -> Result { - // The resolved name is the upstream model id (per the crate contract). - llm_request.model = Some(model.to_string()); let mut body = encode_request(&llm_request, wire_format) .map_err(|error| LlmClientError::RequestEncoding(error.to_string()))?; // `encode_request` round-trips a preserved same-format body verbatim, @@ -365,35 +364,34 @@ impl TranslatingLlmClient { request: Request, model_name: Option<&ModelId>, ) -> Result { - // Own the request's parts so the model can be set without a `mut` param - // and without cloning the messages. `raw_request` is unused here. let Request { - llm_request, + mut llm_request, metadata, .. } = request; - let model = model_name + let model_id = model_name .cloned() - .or_else(|| llm_request.model.clone().map(ModelId::from)) + .or_else(|| llm_request.model.map(ModelId::from)) .ok_or_else(|| LlmClientError::InvalidRequest { message: "no model given".to_string(), })?; + llm_request.model = Some(model_id.to_string()); let orig_format = metadata.as_ref().and_then(|m| m.wire_format); let wire_format = orig_format.unwrap_or( self.model_to_config - .get(&model) + .get(&model_id) .map(|config| config.default_backend.wire_format()) .ok_or_else(|| LlmClientError::Configuration { - message: format!("no backend configured for model {model:?}"), + message: format!("no backend configured for model {model_id:?}"), })?, ); - let backend = - self.backend_for(&model, wire_format) - .ok_or_else(|| LlmClientError::Configuration { - message: format!("model {model:?} has no backend for format {wire_format}"), - })?; + let backend = self.backend_for(&model_id, wire_format).ok_or_else(|| { + LlmClientError::Configuration { + message: format!("model {model_id:?} has no backend for format {wire_format}"), + } + })?; let http_response = self .send_encoded( @@ -401,7 +399,7 @@ impl TranslatingLlmClient { wire_format, llm_request, metadata.as_ref(), - &model, + &model_id, UpstreamEndpoint::Completion, ) .await?; @@ -505,9 +503,8 @@ impl TranslatingLlmClient { #[async_trait] impl RoutedLlmClient for TranslatingLlmClient { - async fn call(&self, request: Request, decision: Decision) -> Result { - let model_name = Some(decision.selected_model_id()); - self.call_rewrite_model(request, model_name).await + async fn call(&self, request: Request) -> Result { + self.call_rewrite_model(request, None).await } } @@ -1710,9 +1707,8 @@ mod tests { client.client = reqwest::Client::builder() .timeout(std::time::Duration::from_millis(10)) .build()?; - let decision = fixed_decision("gpt"); - let Err(error) = client.call(request_for(None, false), decision).await else { + let Err(error) = client.call(request_for(Some("gpt"), false)).await else { panic!("expected a timeout"); }; let LlmClientError::Timeout { source } = error else { @@ -1809,14 +1805,10 @@ mod tests { Ok(()) } - fn fixed_decision(target: &str) -> Decision { - Decision::new(target, None, true) - } - - // Exercises the `RoutedLlmClient` impl: `call` resolves the upstream model from the - // decision (the request carries none) and round-trips a buffered response. + // Exercises the `RoutedLlmClient` impl: `call` uses the model already materialized in the + // request and round-trips a buffered response. #[tokio::test] - async fn routed_llm_client_serves_the_decision_model() + async fn routed_llm_client_serves_the_request_model() -> std::result::Result<(), Box> { let server = MockServer::start().await; Mock::given(method("POST")) @@ -1835,9 +1827,7 @@ mod tests { .await; let client = TranslatingLlmClient::new(&chat_map(&format!("{}/v1", server.uri())))?; - let decision = fixed_decision("gpt"); - // Called through the trait; the request has no model, so "gpt" comes from the decision. - let response = client.call(request_for(None, false), decision).await?; + let response = client.call(request_for(Some("gpt"), false)).await?; let agg = response.llm_response.into_agg().await?; assert_eq!(completion_text(&agg), "routed hi"); Ok(()) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 0324035a..1bcf6624 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -109,12 +109,12 @@ impl RoutedCallWindows { fields( algorithm = call.algorithm, switchyard.algorithm = call.algorithm, - selected_model = %call.decision.selected_model_id(), + selected_model = call.selected_model_id(), otel.kind = "client", - otel.name = %format_args!("chat {}", call.decision.selected_model_id()), + otel.name = %format_args!("chat {}", call.selected_model_id()), openinference.span.kind = "LLM", gen_ai.operation.name = "chat", - gen_ai.request.model = %call.decision.selected_model_id(), + gen_ai.request.model = call.selected_model_id(), gen_ai.request.stream = tracing::field::Empty, gen_ai.request.temperature = tracing::field::Empty, gen_ai.request.top_p = tracing::field::Empty, @@ -155,26 +155,25 @@ async fn serve( { span.record("gen_ai.conversation.id", session_id); } + let target = ModelId::from(call.selected_model_id()); let request = call.request.clone(); - let decision = call.decision.clone(); - let target = decision.selected_model_id().clone(); - let is_answer_call = decision.is_answer_call(); + let is_answer_call = call.decision.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(&target); let started = Instant::now(); let result = match client { - Ok(client) => client.call(request, decision).await, + Ok(client) => client.call(request).await, Err(error) => Err(error), } - .map_err(|source| LibsyError::client_call(target, source)); + .map_err(|source| LibsyError::client_call(target.clone(), source)); let ended = Instant::now(); let duration = ended - started; let result = observability::observe_client_call(result); if let Some(observer) = observer { observer(RunObservation::LlmCall(LlmCallObservation { - selected_model: call.decision.selected_model_id().clone(), + selected_model: target, is_answer_call, is_success: result.is_ok(), duration, diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index d6f34be9..9f13e55c 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -327,19 +327,16 @@ struct UsageClient { /// Client that returns a weak classifier verdict. The delays let a test tell /// classifier time apart from routed-call time. struct ClassifierClient { + classifier_model_id: ModelId, classifier_delay: Duration, routed_delay: Duration, } #[async_trait] impl RoutedLlmClient for ClassifierClient { - async fn call( - &self, - _request: Request, - decision: Decision, - ) -> Result { - let model = decision.selected_model_id().to_string(); - let completion = if decision.is_answer_call() { + async fn call(&self, request: Request) -> Result { + let model_id = request.model_id().unwrap_or_default(); + let completion = if model_id != self.classifier_model_id { tokio::time::sleep(self.routed_delay).await; "routed response" } else { @@ -347,7 +344,7 @@ impl RoutedLlmClient for ClassifierClient { r#"{"crux":"bounded task","primary_rule":"SUP-1","capability_boundary":"supported","p_solve":0.9}"# }; Ok(Response { - llm_response: LlmResponse::Agg(text_response(Some(model), completion)), + llm_response: LlmResponse::Agg(text_response(Some(model_id.to_string()), completion)), metadata: None, }) } @@ -361,20 +358,17 @@ enum JudgeOutcome { /// Returns one configured judge outcome and serves the selected target normally. struct JudgeClient { + judge_model: ModelId, outcome: JudgeOutcome, } #[async_trait] impl RoutedLlmClient for JudgeClient { - async fn call( - &self, - _request: Request, - decision: Decision, - ) -> Result { - if decision.is_answer_call() { + async fn call(&self, request: Request) -> Result { + if request.model_id().as_deref().unwrap_or_default() != self.judge_model { return Ok(Response { llm_response: LlmResponse::Agg(text_response( - Some(decision.selected_model_id().to_string()), + request.model_id().map(|id| id.to_string()), "routed response", )), metadata: None, @@ -408,11 +402,10 @@ impl RoutedLlmClient for JudgeClient { impl RoutedLlmClient for UsageClient { async fn call( &self, - _request: Request, - decision: Decision, + request: Request, ) -> Result { let mut response = text_response( - Some(decision.selected_model_id().to_string()), + request.model_id().map(|s| s.to_string()), "observed response", ); response.id = Some("obs-response-1".to_string()); @@ -967,11 +960,7 @@ struct StreamingUsageClient; #[async_trait] impl RoutedLlmClient for StreamingUsageClient { - async fn call( - &self, - _request: Request, - decision: Decision, - ) -> Result { + async fn call(&self, request: Request) -> Result { let usage = Usage { input_tokens: Some(13), output_tokens: Some(5), @@ -981,7 +970,7 @@ impl RoutedLlmClient for StreamingUsageClient { let chunks = vec![Ok(LlmResponseStreamEvent::new(vec![ LlmResponseChunk::MessageStart { id: Some("obs-stream-response".to_string()), - model: Some(decision.selected_model_id().to_string()), + model: request.model_id().map(|s| s.to_string()), }, LlmResponseChunk::Usage(usage), LlmResponseChunk::MessageStop { @@ -999,11 +988,7 @@ struct TimeoutClient; #[async_trait] impl RoutedLlmClient for TimeoutClient { - async fn call( - &self, - _request: Request, - _decision: Decision, - ) -> Result { + async fn call(&self, _request: Request) -> Result { Err(LlmClientError::Timeout { source: Box::new(TestError("upstream timed out")), }) @@ -1236,6 +1221,7 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib u64_gauge_value(&before, "switchyard.total_requests").unwrap_or_default(); let client = Arc::new(ClassifierClient { + classifier_model_id: "classifier".into(), classifier_delay: Duration::from_millis(60), routed_delay: Duration::from_millis(200), }) as Arc; @@ -1329,7 +1315,10 @@ async fn classifier_fail_open_records_each_failure_stage() -> switchyard_libsy:: ]; for (judge_model, outcome, expected_reason) in cases { - let client = Arc::new(JudgeClient { outcome }) as Arc; + let client = Arc::new(JudgeClient { + judge_model: judge_model.into(), + outcome, + }) as Arc; run( classifier_router(judge_model, "fo-weak", "fo-strong")?, client, diff --git a/crates/libsy/src/algorithms/noop.rs b/crates/libsy/src/algorithms/noop.rs index 1ffc3248..acf24097 100644 --- a/crates/libsy/src/algorithms/noop.rs +++ b/crates/libsy/src/algorithms/noop.rs @@ -6,7 +6,8 @@ use std::sync::Arc; use switchyard_protocol::{ - AggLlmResponse, ContentBlock, LlmResponse, Request, Response, ResponseOutput, Role, StopReason, + AggLlmResponse, ContentBlock, LlmResponse, ModelId, Request, Response, ResponseOutput, Role, + StopReason, }; use crate::Result; @@ -23,12 +24,11 @@ impl Algorithm for Noop { } async fn route(self: Arc, driver: Driver, request: Request) -> Result { - let model = request - .requested_model() - .unwrap_or("switchyard/noop") - .to_string(); + let model_id = request + .model_id() + .unwrap_or_else(|| ModelId::from("switchyard/noop")); let decision: Decision = Decision::new( - model.clone(), + model_id.clone(), Some("noop returned its synthetic response".to_string()), true, ); @@ -36,7 +36,7 @@ impl Algorithm for Noop { let llm_response = LlmResponse::Agg(AggLlmResponse { id: Some("switchyard-noop".to_string()), - model: Some(model), + model: Some(model_id.to_string()), outputs: vec![ResponseOutput { role: Role::Assistant, content: vec![ContentBlock::Text { diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 6df5c98e..12947486 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -47,18 +47,16 @@ pub type StepStream = Pin> + Send>>; /// 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. /// -/// The selected model and inbound route name live in separate, unambiguous places: the model -/// identifier is [`decision.selected_model_id()`](Decision::selected_model_id), while -/// `request.llm_request.model` is the *inbound* name the agent asked for (libsy -/// never overwrites it). A client maps `selected_model_id()` to the provider model -/// id it hits. +/// The selected model is available both from +/// [`decision.selected_model_id()`](Decision::selected_model_id) and from +/// `request.llm_request.model`. [`Driver::call_model`] stamps the decision's model onto the +/// request before publishing the call, so every consumer receives a request ready for the +/// selected target. pub struct CallModel { /// The name of the algorithm that produced this call, so a host instrumenting the /// calls it serves can attribute its own spans to the algorithm behind them. pub algorithm: String, - /// The request to serve; its `model` is the agent's original name NOT the selected model. - /// The caller making the request needs to change it to decision.selected_model_id() before - /// sending. + /// The request to serve; its `model` is the selected model identified by `decision`. pub request: Request, /// The routing decision behind this call; `selected_model_id()` identifies the model to use. pub decision: Decision, @@ -67,6 +65,14 @@ pub struct CallModel { } impl CallModel { + /// The selected model stamped onto this call's request by [`Driver::call_model`]. + pub fn selected_model_id(&self) -> &str { + let model_id = self.request.llm_request.model.as_deref(); + // Driver stamps the model before constructing CallModel. + debug_assert!(model_id.is_some()); + model_id.unwrap_or_default() + } + /// Fulfill the promise with the caller's model-call result. Pass `Err(..)` to /// propagate a failed model call back to the algorithm. Consumes the promise: it /// can only be fulfilled once. @@ -151,8 +157,9 @@ impl Driver { reasoning_tokens = tracing::field::Empty, ) )] - pub async fn call_model(&self, request: Request, decision: Decision) -> Result { + pub async fn call_model(&self, mut request: Request, decision: Decision) -> Result { let selected_model_id = decision.selected_model_id().to_string(); + request.llm_request.model = Some(selected_model_id.clone()); let is_answer_call = decision.is_answer_call(); let started = Instant::now(); let (reply, response) = oneshot::channel::>(); @@ -721,7 +728,8 @@ mod tests { let Step::CallModel(call) = step else { return Err(test_error("expected a CallModel step")); }; - calls.insert(call.decision.selected_model_id().to_string(), call); + let selected_model = call.selected_model_id().to_string(); + calls.insert(selected_model, call); } assert!( tokio::time::timeout(std::time::Duration::from_millis(20), &mut first) @@ -793,7 +801,7 @@ mod tests { /// 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_call_without_answering_it() -> Result<()> { + async fn into_parts_yields_the_selected_model_without_answering_it() -> Result<()> { let (driver, mut step_rx) = Driver::new("test"); let decision = test_decision(ModelId::from("answer/model")); let producer = tokio::spawn({ @@ -808,7 +816,13 @@ mod tests { let (taken_request, taken_decision) = call.into_parts(); assert_eq!(taken_decision.selected_model_id(), "answer/model"); assert!(taken_decision.is_answer_call()); - assert_eq!(taken_request.llm_request, request().llm_request); + 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 @@ -929,8 +943,7 @@ mod tests { match step? { Step::CallModel(call) => { saw_call = true; - // The decision rode along with the promise. - assert_eq!(call.decision.selected_model_id(), "offload/model"); + assert_eq!(call.selected_model_id(), "offload/model"); // Fulfilling the promise is the "real" model call the caller makes. call.respond(Ok(Response { llm_response: LlmResponse::Agg(text_response( diff --git a/crates/libsy/src/core/classifier.rs b/crates/libsy/src/core/classifier.rs index f6212f67..77191ae0 100644 --- a/crates/libsy/src/core/classifier.rs +++ b/crates/libsy/src/core/classifier.rs @@ -193,10 +193,10 @@ mod tests { _driver: Option<&Driver>, ) -> Result<(Classification, Option)> { *state = true; - let target = request.requested_model().unwrap_or("auto").to_string(); + let target = request.model_id().unwrap_or(ModelId::from("auto")); Ok(( Classification::Scores(vec![Score { - target: target.into(), + target, confidence: 1.0, }]), None, @@ -261,7 +261,7 @@ mod tests { // The rewrite outlives the call: later classifiers in the cascade score this value, // and it is what reaches the model. - assert_eq!(request.requested_model(), Some("rewritten")); + assert_eq!(request.model_id().as_deref(), Some("rewritten")); Ok(()) } } diff --git a/crates/libsy/src/core/processor.rs b/crates/libsy/src/core/processor.rs index 7f23e7d0..b84727bb 100644 --- a/crates/libsy/src/core/processor.rs +++ b/crates/libsy/src/core/processor.rs @@ -141,14 +141,14 @@ mod tests { async fn processor_rewrites_the_request_in_place() -> Result<()> { let mut state = (); let mut req = request(); - assert_eq!(req.requested_model(), Some("auto")); + assert_eq!(req.model_id(), Some("auto".into())); RewritingProcessor .process(&mut state, Event::Request(&mut req)) .await?; // The edit outlives the call, so the next component sees the rewritten request. - assert_eq!(req.requested_model(), Some("rewritten")); + assert_eq!(req.model_id(), Some("rewritten".into())); Ok(()) } } diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 2435cb02..36d39c4e 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -114,8 +114,8 @@ pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span { outcome = tracing::field::Empty, error = tracing::field::Empty, ); - if let Some(route) = request.requested_model() { - span.record("switchyard.route", route); + if let Some(route) = request.model_id() { + span.record("switchyard.route", route.as_ref()); } if let Some(metadata) = &request.metadata { for (field, value) in [ diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index 21f2ac2e..3b758503 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -180,10 +180,6 @@ impl Decision { /// not serialize requests unless their transport requires it. #[async_trait] pub trait RoutedLlmClient: Send + Sync { - /// Serve the model identified by - /// [`decision.selected_model_id()`](Decision::selected_model_id), resolving it to the - /// provider model this client calls. - /// `request.llm_request.model` is the agent's original name, carried through for - /// reference, not a call target. - async fn call(&self, request: Request, decision: Decision) -> Result; + /// Make a request + async fn call(&self, request: Request) -> Result; } diff --git a/crates/protocol/src/envelope.rs b/crates/protocol/src/envelope.rs index 227a9d4c..f4c4b257 100644 --- a/crates/protocol/src/envelope.rs +++ b/crates/protocol/src/envelope.rs @@ -4,7 +4,7 @@ //! The request/response envelope: the normalized [`LlmRequest`]/[`LlmResponse`] paired //! with the original provider payload and correlation [`Metadata`]. -use crate::{LlmRequest, LlmResponse, Metadata}; +use crate::{LlmRequest, LlmResponse, Metadata, ModelId}; /// A request an algorithm routes: the normalized [`LlmRequest`] plus optional /// host-owned raw data and correlation [`Metadata`]. @@ -21,11 +21,12 @@ pub struct Request { } impl Request { - /// Returns the model name supplied by the inbound request, when present. + /// Returns the model currently carried by the request, when present. /// - /// This is not necessarily the target selected by a routing decision. - pub fn requested_model(&self) -> Option<&str> { - self.llm_request.model.as_deref() + /// On ingress this is the name supplied by the client. Before an offloaded model call, the + /// routing driver replaces it with the selected target. + pub fn model_id(&self) -> Option { + self.llm_request.model.as_deref().map(ModelId::from) } } diff --git a/crates/protocol/src/llm.rs b/crates/protocol/src/llm.rs index d8def175..c2386c23 100644 --- a/crates/protocol/src/llm.rs +++ b/crates/protocol/src/llm.rs @@ -300,7 +300,10 @@ pub struct PreservationMetadata { #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] #[serde(default)] pub struct LlmRequest { - /// Model requested by the inbound client. + /// Model currently addressed by the request. + /// + /// This initially contains the name supplied by the inbound client. A routing host may + /// replace it with the selected target before serving the request. pub model: Option, /// System and developer instructions separated from conversation turns. pub instructions: Vec, diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 2bed6b40..d7ca4039 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -18,7 +18,7 @@ use switchyard_libsy::{ }; use switchyard_llm_client::ClientRouter; use switchyard_protocol::{ - AggLlmResponse, Decision, LlmClientError, LlmResponse, Metadata, ModelId, Request, Response, + AggLlmResponse, LlmClientError, LlmResponse, Metadata, ModelId, Request, Response, RoutedLlmClient, }; @@ -47,11 +47,7 @@ struct PythonLlmClient { #[async_trait] impl RoutedLlmClient for PythonLlmClient { - async fn call( - &self, - request: Request, - _decision: Decision, - ) -> Result { + async fn call(&self, request: Request) -> Result { let metadata = request.metadata; let future = Python::attach(|py| { let request = to_python(py, &request.llm_request)?; diff --git a/tests/test_libsy_minimal_bindings.py b/tests/test_libsy_minimal_bindings.py index 8067e3ff..059ccf0a 100644 --- a/tests/test_libsy_minimal_bindings.py +++ b/tests/test_libsy_minimal_bindings.py @@ -54,6 +54,7 @@ async def test_random_runs_with_a_python_client() -> None: "is_answer_call": True, } ] + assert client.calls[0]["model"] == "fast" assert client.calls[0]["messages"][0]["content"] == [ {"type": "text", "text": "hello"} ]