From 00e9f14ca96e0ff7776ae3a153cb4b5badf359f8 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Fri, 2 Oct 2026 14:33:33 -0700 Subject: [PATCH] feat(client): serve System One decision calls Signed-off-by: nachiketb --- Cargo.lock | 1 + crates/libsy-llm-client/Cargo.toml | 1 + crates/libsy-llm-client/src/client.rs | 4 +- crates/libsy-llm-client/src/lib.rs | 7 +- crates/libsy-llm-client/src/observation.rs | 9 +- crates/libsy-llm-client/src/run.rs | 276 ++++++++++++++++-- crates/libsy-llm-client/src/system_one.rs | 242 +++++++++++++++ crates/protocol/src/client.rs | 18 +- .../src/runtime.rs | 21 +- crates/switchyard-server/src/lib.rs | 2 +- 10 files changed, 543 insertions(+), 38 deletions(-) create mode 100644 crates/libsy-llm-client/src/system_one.rs diff --git a/Cargo.lock b/Cargo.lock index aeeeaf37b..48b15ca63 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2358,6 +2358,7 @@ dependencies = [ "opentelemetry_sdk", "parking_lot", "reqwest", + "serde", "serde_json", "switchyard-libsy", "switchyard-protocol", diff --git a/crates/libsy-llm-client/Cargo.toml b/crates/libsy-llm-client/Cargo.toml index cb8ea4f93..ded0ecab8 100644 --- a/crates/libsy-llm-client/Cargo.toml +++ b/crates/libsy-llm-client/Cargo.toml @@ -29,6 +29,7 @@ parking_lot.workspace = true http.workspace = true httpdate.workspace = true serde_json.workspace = true +serde.workspace = true tokio.workspace = true tracing.workspace = true tracing-opentelemetry.workspace = true diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 6ba089d9c..b66a38b42 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! [`TranslatingLlmClient`] — the crate's single public entry point: encode a neutral +//! [`TranslatingLlmClient`]: encode a neutral //! request, call the configured backend over HTTP, decode the neutral response. use std::collections::{BTreeMap, BTreeSet, HashMap}; @@ -896,7 +896,7 @@ fn record_gen_ai_request(url: &str, model: &str, streaming: bool) { } } -fn convert_reqwest_error(error: reqwest::Error) -> LlmClientError { +pub(crate) fn convert_reqwest_error(error: reqwest::Error) -> LlmClientError { // Reqwest labels truncated or otherwise unreadable response bodies as decode // errors, so distinguish them from serde JSON failures at the call site. let error = error.without_url(); diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index d5f579c7d..db0cf9c0c 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -11,6 +11,9 @@ //! back to a [`switchyard_protocol::Response`] — supporting both buffered and //! streamed responses. //! +//! [`SystemOneClient`] serves typed decision requests. Register decision targets +//! with [`ClientRouter::with_decision_clients`] to serve them alongside LLM calls. +//! //! [`run()`] pairs the client with a libsy algorithm: it drives //! [`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 @@ -24,14 +27,16 @@ mod observability; mod observation; pub mod raw; pub mod run; +mod system_one; pub use backend::{Backend, DEFAULT_MAX_RETRIES, HttpBackendConfig}; pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; -pub use observation::{LlmCallObservation, RunObservation, RunObserver}; +pub use observation::{LlmCallObservation, ModelCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; pub use run::{ClientRouter, decide, run}; pub use switchyard_translation::RawEventStream; +pub use system_one::SystemOneClient; /// Registers process-wide compatibility gauges with the global meter provider. pub fn initialize_metrics() { diff --git a/crates/libsy-llm-client/src/observation.rs b/crates/libsy-llm-client/src/observation.rs index 308ad4b28..0fc97a5e5 100644 --- a/crates/libsy-llm-client/src/observation.rs +++ b/crates/libsy-llm-client/src/observation.rs @@ -11,7 +11,7 @@ use switchyard_protocol::{ModelId, Usage}; /// One completed model call observed while serving an algorithm run. #[derive(Clone, Debug)] -pub struct LlmCallObservation { +pub struct ModelCallObservation { /// Model selected for the completed call. pub selected_model: ModelId, /// Whether the call completed successfully. @@ -22,13 +22,18 @@ pub struct LlmCallObservation { pub usage: Option, } -/// Events emitted inline while [`crate::run`] serves a routing request. +/// An LLM call's observation, sharing the fields used by decision calls. +pub type LlmCallObservation = ModelCallObservation; + +/// Events emitted inline while [`crate::run()`] serves a routing request. #[derive(Clone, Debug)] pub enum RunObservation { /// Metadata attached to the completed routing outcome. Outcome(OutcomeMetadata), /// A completed model call requested by the algorithm for routing work. LlmCall(LlmCallObservation), + /// A completed decision call requested by the algorithm for routing work. + DecisionCall(ModelCallObservation), /// A completed terminal model call made from the routing outcome. AnswerCall(LlmCallObservation), /// Routing time recorded by the `switchyard.routing_overhead_ms` metric. diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index a3dfc339a..f694fa05c 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -32,12 +32,12 @@ use switchyard_libsy::{ }; use switchyard_protocol::{ AggLlmResponse, LlmClientError, LlmResponse, LlmResponseChunk, LlmResponseStream, Message, - ModelId, Request, Response, ResponseAccumulator, RoutedLlmClient, RoutingFallbackReason, - WireFormat, + ModelId, Request, Response, ResponseAccumulator, RoutedDecisionClient, RoutedLlmClient, + RoutingFallbackReason, WireFormat, }; use switchyard_translation::prepare_request_for_target; -use crate::observation::{LlmCallObservation, RunObservation, RunObserver}; +use crate::observation::{LlmCallObservation, ModelCallObservation, RunObservation, RunObserver}; use crate::{metrics, observability}; /// Run one request to completion, serving every offloaded model call with `client`. @@ -152,18 +152,38 @@ pub async fn decide( Ok(outcome) } -async fn unsupported_decision(call: CallDecision) -> Result<()> { - let model = call.model.clone(); - call.respond(Err(LibsyError::client_call( - model, - LlmClientError::General("decision calls are not supported by this client".to_string()), - ))) +#[tracing::instrument( + name = "libsy.decision_client_call", + skip_all, + fields(algorithm = %call.algorithm, selected_model = %call.model, outcome = tracing::field::Empty), +)] +async fn serve_decision( + clients: &ClientRouter, + call: CallDecision, + observations: &Option>>>, +) -> Result<()> { + let client = clients.route_decision(&call.model); + let started = Instant::now(); + let result = async { client?.call(call.request.clone()).await }.await; + tracing::Span::current().record("outcome", if result.is_ok() { "ok" } else { "error" }); + if let Some(observations) = observations { + observations + .lock() + .push(RunObservation::DecisionCall(ModelCallObservation { + selected_model: call.model.clone(), + is_success: result.is_ok(), + duration: started.elapsed(), + usage: result.as_ref().ok().map(|response| response.usage.clone()), + })); + } + let result = result.map_err(|error| LibsyError::client_call(call.model.clone(), error)); + call.respond(result) } /// Emits completed routing calls after the outcome reveals whether one response became the answer. fn emit_routing_observations( observer: &Option, - observations: &Option>>>, + observations: &Option>>>, answered_model: Option<&ModelId>, ) { let (Some(observer), Some(observations)) = (observer, observations) else { @@ -171,31 +191,36 @@ fn emit_routing_observations( }; let mut answer_observed = false; for observation in observations.lock().drain(..) { - if !answer_observed && answered_model == Some(&observation.selected_model) { - answer_observed = true; - observer(RunObservation::AnswerCall(observation)); - } else { - observer(RunObservation::LlmCall(observation)); - } + observer(match observation { + RunObservation::LlmCall(call) + if !answer_observed && answered_model == Some(&call.selected_model) => + { + answer_observed = true; + RunObservation::AnswerCall(call) + } + observation => observation, + }); } } /// Serve one offloaded call and fulfill its promise. /// -/// LLM failures stop the run unless the call enables recovery. Unsupported decisions -/// return an error to the algorithm. +/// LLM failures stop the run unless the call enables recovery. Decision results, +/// including client failures, go back to the algorithm for its routing policy. async fn serve( clients: ClientRouter, call: Call, - observations: Option>>>, + observations: Option>>>, ) -> Result<()> { let call = match call { Call::Model(call) => *call, - Call::Decision(call) => return unsupported_decision(*call).await, + Call::Decision(call) => return serve_decision(&clients, *call, &observations).await, }; let observe = |observation| { if let Some(observations) = &observations { - observations.lock().push(observation); + observations + .lock() + .push(RunObservation::LlmCall(observation)); } }; let target = call.models.first().ok_or(LibsyError::NoTargets)?; @@ -557,13 +582,13 @@ fn conversation_id(fields: &serde_json::Map) -> Option<&str> { /// /// An algorithm routes among named targets; which provider each target lives on is the /// host's concern, and two targets in one run may sit on different providers. A router owns -/// that mapping. It is *not* itself a client: it hands back a [`RoutedLlmClient`] and the -/// caller makes the call. +/// that mapping and resolves the client for each kind of call. /// /// Cloning is cheap — the mapping is shared, so one router can serve every request. #[derive(Clone)] pub struct ClientRouter { inner: Arc, + decision_clients: Arc>>, } struct ClientRouting { @@ -613,7 +638,7 @@ impl ClientRouter { /// Build a router with an explicit list of models that can answer requests. /// - /// `by_model` contains every callable model, including classifiers. Only the model IDs in + /// `by_model` contains callable LLMs, including LLM classifiers. Only the model IDs in /// `completion_targets` determine whether native Responses state needs a routing pin. /// Cross-format state is recorded lazily when a Responses request uses a Chat or Anthropic /// target. `target_prompts` and `routing_answer_target` have the same meaning as in @@ -631,6 +656,7 @@ impl ClientRouter { .next() .is_some_and(|first| clients.any(|client| !Arc::ptr_eq(first, client))); Self { + decision_clients: Arc::default(), inner: Arc::new(ClientRouting { routing: Routing::ByModel(by_model), target_prompts, @@ -648,6 +674,7 @@ impl ClientRouter { /// only duplicate that. pub fn single(client: Arc) -> Self { Self { + decision_clients: Arc::default(), inner: Arc::new(ClientRouting { routing: Routing::Single(client), target_prompts: HashMap::new(), @@ -678,6 +705,27 @@ impl ClientRouter { } } + /// Register clients for decision targets independently of completion targets. + pub fn with_decision_clients( + mut self, + clients: HashMap>, + ) -> Self { + self.decision_clients = Arc::new(clients); + self + } + + /// Resolve a decision target without falling back to an LLM client. + pub fn route_decision( + &self, + model: &ModelId, + ) -> std::result::Result<&Arc, LlmClientError> { + self.decision_clients + .get(model) + .ok_or_else(|| LlmClientError::Configuration { + message: format!("no decision client is configured for model {model:?}"), + }) + } + /// Return the recorded model for the requested response or conversation ID. fn stored_state_owner(&self, request: &Request) -> Option { let fields = &request.llm_request.extensions.fields; @@ -1072,6 +1120,185 @@ mod tests { } } + #[tokio::test] + async fn system_one_serves_decisions_and_returns_failures_to_the_classifier() + -> std::result::Result<(), Box> { + use std::time::Duration; + use switchyard_libsy::{ + CapabilityJudgeConfig, DecisionJudgeConfig, LlmClassifierConfig, LlmTaskClassifier, + TaskClassifierConfig, + }; + use switchyard_protocol::{ + BooleanEstimate, DecisionRequest, DecisionValue, Probability, ScoreValue, + }; + use wiremock::matchers::{body_partial_json, header, path}; + + let server = MockServer::start().await; + let client = Arc::new(crate::SystemOneClient::new( + format!("{}/v1/systemone", server.uri()), + "test-key".into(), + Duration::from_secs(2), + )?); + let mixed: DecisionRequest = serde_json::from_value(json!({ + "model": "judge", "context": {"task": "A simple task"}, + "questions": { + "boolean": {"instructions": "Is it simple?", "kind": {"type": "boolean", "data": { + "true_description": ["Simple"], "false_description": null + }}}, + "route": {"instructions": {"task": "Choose"}, "kind": {"type": "choice", "data": { + "options": [{"id": "advantage", "description": "A wins"}, {"id": "no_advantage"}] + }}}, + "score": {"instructions": "Rate difficulty", "kind": {"type": "score", "data": { + "levels": ["Easy", {"description": "Hard"}] + }}} + } + }))?; + let mut body = json!({ + "model": "jev-1.13.0", "usage": {"input_tokens": 42, "output_tokens": 3}, + "answers": { + "boolean": {"type": "noul", "noul": 0.9}, + "route": {"type": "choice", "choice": "no_advantage", "confidence": 0.7, + "probabilities": {"advantage": 0.2, "no_advantage": 0.8}}, + "score": {"type": "score", "score": 0.6, "confidence": 0.3, + "legend": {"0": "Easy", "1": {"description": "Hard"}}, + "probabilities": {"1": 0.6, "0": 0.4}} + } + }); + Mock::given(method("POST")) + .and(path("/v1/systemone")) + .and(header("authorization", "Bearer test-key")) + .and(body_partial_json(json!({ + "model": "judge", "state": mixed.context, + "questions": { + "boolean": {"type": "noul", "criteria": {"true": ["Simple"]}}, + "route": {"type": "choice", "instructions": {"task": "Choose"}, + "criteria": {"advantage": "A wins", "no_advantage": null}}, + "score": {"type": "score", "criteria": ["Easy", {"description": "Hard"}]} + } + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(&body)) + .expect(1) + .mount(&server) + .await; + let response = client.call(mixed).await?; + assert_eq!(response.model.as_deref(), Some("jev-1.13.0")); + assert_eq!(response.id, None); + assert_eq!(response.usage.input_tokens, Some(42)); + assert_eq!(response.usage.total_tokens, None); + assert_eq!( + response.answers["boolean"].value, + DecisionValue::Boolean(BooleanEstimate::ProbabilityTrue(Probability(0.9))) + ); + assert_eq!( + response.answers["score"].value, + DecisionValue::Score { + value: ScoreValue(0.6), + probabilities: Some(vec![Probability(0.4), Probability(0.6)]), + } + ); + assert_eq!( + response.answers["route"].provider_confidence.map(|c| c.0), + Some(0.7) + ); + server.verify().await; + + body["answers"].as_object_mut().unwrap().remove("boolean"); + body["answers"].as_object_mut().unwrap().remove("score"); + let models = Arc::new(RuntimeModels::new(HashMap::from([ + (Category::Judge, vec!["judge".into()]), + (Category::Capable, vec!["capable".into()]), + (Category::Efficient, vec!["efficient".into()]), + (Category::Any, vec!["efficient".into(), "capable".into()]), + ]))); + let algorithm = || -> Result> { + Ok(Arc::new(LlmTaskClassifier::new( + LlmClassifierConfig::Capability { + config: TaskClassifierConfig { + judge: CapabilityJudgeConfig::Decision(DecisionJudgeConfig { + cutoff: 0.4, + instructions: None, + candidates: BTreeMap::from([ + ("a".into(), "capable".into()), + ("b".into(), "efficient".into()), + ]), + evidence: json!({}), + }), + ..TaskClassifierConfig::default() + }, + }, + )?)) + }; + for (template, expected, is_success) in [ + ( + ResponseTemplate::new(200).set_body_json(&body), + "efficient", + true, + ), + ( + ResponseTemplate::new(200).set_body_string("invalid JSON"), + "capable", + false, + ), + ( + ResponseTemplate::new(503).set_body_string("unavailable"), + "capable", + false, + ), + ] { + server.reset().await; + Mock::given(method("POST")) + .and(path("/v1/systemone")) + .respond_with(template) + .expect(2) + .mount(&server) + .await; + let llm = Arc::new(CandidateClient { + calls: Mutex::default(), + requests: Mutex::default(), + first: FirstOutcome::Unauthorized, + }); + let clients = + ClientRouter::single(llm.clone()).with_decision_clients(HashMap::from([( + ModelId::from("judge"), + client.clone() as Arc, + )])); + let outcome = decide(algorithm()?, clients.clone(), request(), models.clone()).await?; + assert_eq!(outcome.selected_model_id()?, expected); + assert!(llm.calls.lock().is_empty()); + let events = Arc::new(Mutex::new(Vec::new())); + let captured = events.clone(); + let (selected, _) = run( + algorithm()?, + clients, + request(), + models.clone(), + Some(Arc::new(move |event| captured.lock().push(event))), + ) + .await?; + assert_eq!(selected, expected); + assert_eq!(&*llm.calls.lock(), &[ModelId::from(expected)]); + { + let events = events.lock(); + let RunObservation::DecisionCall(call) = &events[0] else { + panic!("missing decision observation") + }; + assert_eq!(call.selected_model, "judge"); + assert_eq!(call.is_success, is_success); + assert_eq!( + call.usage.as_ref().and_then(|u| u.input_tokens), + is_success.then_some(42) + ); + assert!( + events + .iter() + .any(|event| matches!(event, RunObservation::AnswerCall(_))) + ); + } + server.verify().await; + } + Ok(()) + } + fn instruction_text(request: &Request) -> Vec<&str> { request .llm_request @@ -1165,6 +1392,7 @@ mod tests { duration: std::time::Duration::ZERO, usage: None, }) + .map(RunObservation::LlmCall) .into(), ))); let emitted = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/libsy-llm-client/src/system_one.rs b/crates/libsy-llm-client/src/system_one.rs new file mode 100644 index 000000000..caf4881e5 --- /dev/null +++ b/crates/libsy-llm-client/src/system_one.rs @@ -0,0 +1,242 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Buffered System One calls using the provider-neutral decision IR. + +use std::collections::BTreeMap; +use std::time::Duration; + +use async_trait::async_trait; +use serde::Deserialize; +use serde_json::{Value, json}; +use switchyard_protocol::{ + BooleanEstimate, DecisionAnswer, DecisionKind, DecisionRequest, DecisionResponse, + DecisionValue, ModelId, Probability, ProviderConfidence, RoutedDecisionClient, ScoreValue, + Usage, +}; + +use crate::client::convert_reqwest_error; +use crate::{LlmClientError, Result}; + +/// Serves decision requests through a System One endpoint, such as TypeSafe's Jev API. +pub struct SystemOneClient { + client: reqwest::Client, + endpoint: String, + api_key: String, +} + +impl SystemOneClient { + /// `endpoint` is the full URL, including `/v1/systemone`. Each call makes one + /// attempt; `timeout` covers sending the request and reading the response body. + pub fn new(endpoint: impl Into, api_key: String, timeout: Duration) -> Result { + let client = reqwest::Client::builder() + .timeout(timeout) + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(convert_reqwest_error)?; + Ok(Self { + client, + endpoint: endpoint.into(), + api_key, + }) + } +} + +#[async_trait] +impl RoutedDecisionClient for SystemOneClient { + async fn call(&self, request: DecisionRequest) -> Result { + let response = self + .client + .post(&self.endpoint) + .bearer_auth(&self.api_key) + .json(&encode(&request)?) + .send() + .await + .map_err(convert_reqwest_error)?; + let status = response.status(); + let body = response.bytes().await.map_err(convert_reqwest_error)?; + if !status.is_success() { + return Err(LlmClientError::UpstreamHttp { + status, + body: String::from_utf8_lossy(&body).replace(&self.api_key, "[REDACTED]"), + }); + } + let response: WireResponse = + serde_json::from_slice(&body).map_err(|source| LlmClientError::InvalidResponse { + source: Box::new(source), + })?; + let answers = response + .answers + .into_iter() + .map(|(id, answer)| { + let value = match answer.value { + WireValue::Noul { noul } => { + DecisionValue::Boolean(BooleanEstimate::ProbabilityTrue(noul)) + } + WireValue::Choice { + choice, + probabilities, + } => DecisionValue::Choice { + selected: choice, + probabilities, + }, + WireValue::Score { + score, + probabilities, + } => { + let probabilities = probabilities + .map(|mut probabilities| { + let Some(DecisionKind::Score { levels }) = + request.questions.get(&id).map(|q| &q.kind) + else { + return Err(LlmClientError::ResponseTranslation(format!( + "score answer {id:?} has no matching score question" + ))); + }; + let invalid_rubric = || { + LlmClientError::ResponseTranslation(format!( + "score probabilities for {id:?} do not match its rubric" + )) + }; + if probabilities.len() != levels.len() { + return Err(invalid_rubric()); + } + // JSON keys are strings; order probabilities by the request's rubric. + (0..levels.len()) + .map(|index| { + probabilities + .remove(&index.to_string()) + .ok_or_else(invalid_rubric) + }) + .collect() + }) + .transpose()?; + DecisionValue::Score { + value: score, + probabilities, + } + } + }; + Ok(( + id, + DecisionAnswer { + value, + provider_confidence: answer.confidence, + }, + )) + }) + .collect::>()?; + Ok(DecisionResponse { + id: response.id, + model: response.model, + answers, + usage: response.usage, + }) + } +} + +fn encode(request: &DecisionRequest) -> Result { + let model = request + .model + .as_ref() + .ok_or_else(|| LlmClientError::InvalidRequest { + message: "System One requires a selected model".into(), + })?; + content(&request.context, false, "context")?; + let mut questions = serde_json::Map::new(); + for (id, question) in &request.questions { + let path = format!("questions.{id}"); + content( + &question.instructions, + true, + &format!("{path}.instructions"), + )?; + let (kind, criteria) = match &question.kind { + DecisionKind::Boolean { + true_description, + false_description, + } => { + let mut criteria = serde_json::Map::new(); + for (key, description) in [("true", true_description), ("false", false_description)] + { + if let Some(description) = description { + content(description, true, &format!("{path}.{key}_description"))?; + criteria.insert(key.into(), description.clone()); + } + } + ("noul", Value::Object(criteria)) + } + DecisionKind::Choice { options } => { + let mut criteria = serde_json::Map::new(); + for option in options { + let description = option.description.as_ref().unwrap_or(&Value::Null); + content(description, true, &format!("{path}.options.{}", option.id))?; + if criteria + .insert(option.id.clone(), description.clone()) + .is_some() + { + return Err(LlmClientError::InvalidRequest { + message: format!("duplicate option {:?} in {path}", option.id), + }); + } + } + ("choice", Value::Object(criteria)) + } + DecisionKind::Score { levels } => { + for (index, level) in levels.iter().enumerate() { + content(level, false, &format!("{path}.levels.{index}"))?; + } + ("score", json!(levels)) + } + }; + questions.insert( + id.clone(), + json!({ + "type": kind, "instructions": question.instructions, "criteria": criteria, + }), + ); + } + Ok(json!({"model": model, "state": request.context, "questions": questions})) +} + +fn content(value: &Value, nullable: bool, path: &str) -> Result<()> { + if value.is_string() || value.is_object() || value.is_array() || (nullable && value.is_null()) { + return Ok(()); + } + Err(LlmClientError::RequestEncoding(format!( + "System One cannot represent {path} as {}", + value + ))) +} + +#[derive(Deserialize)] +struct WireResponse { + id: Option, + model: Option, + answers: BTreeMap, + #[serde(default)] + usage: Usage, +} + +#[derive(Deserialize)] +struct WireAnswer { + #[serde(flatten)] + value: WireValue, + confidence: Option, +} + +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum WireValue { + Noul { + noul: Probability, + }, + Choice { + choice: String, + probabilities: Option>, + }, + Score { + score: ScoreValue, + probabilities: Option>, + }, +} diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index 8d312e72d..fa37f01c1 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -1,17 +1,15 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! The routed-call server trait and its shared error types. +//! Routed-call client contracts and 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. 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. +//! Hosts implement these traits to perform model calls. Keeping the contracts here +//! lets clients depend on the protocol without pulling in libsy's orchestration. use async_trait::async_trait; use thiserror::Error; -use crate::{ModelId, Request, Response}; +use crate::{DecisionRequest, DecisionResponse, ModelId, Request, Response}; /// A boxed client-specific error preserved as the source of a routed call failure. pub type BoxError = Box; @@ -157,3 +155,11 @@ pub trait RoutedLlmClient: Send + Sync { /// Make a request async fn call(&self, request: Request) -> Result; } + +/// Serves typed decision calls over a host-owned transport. +/// Implementations may be shared across targets and called concurrently. +#[async_trait] +pub trait RoutedDecisionClient: Send + Sync { + /// Evaluate the request using its selected model. + async fn call(&self, request: DecisionRequest) -> Result; +} diff --git a/crates/switchyard-nemo-relay-plugin/src/runtime.rs b/crates/switchyard-nemo-relay-plugin/src/runtime.rs index 95794f0d8..0a282c3e6 100644 --- a/crates/switchyard-nemo-relay-plugin/src/runtime.rs +++ b/crates/switchyard-nemo-relay-plugin/src/runtime.rs @@ -336,7 +336,23 @@ impl SwitchyardRuntime { } RunObservation::LlmCall(call) => { call_index += 1; - self.routing_call_events(events, call, call_index, metadata); + self.routing_call_events( + events, + call, + call_index, + metadata, + "switchyard.routing.llm_call", + ); + } + RunObservation::DecisionCall(call) => { + call_index += 1; + self.routing_call_events( + events, + call, + call_index, + metadata, + "switchyard.routing.decision_call", + ); } RunObservation::RoutingOverhead(duration) => { let latency_ms = duration.as_secs_f64() * 1_000.0; @@ -377,12 +393,13 @@ impl SwitchyardRuntime { call: LlmCallObservation, call_index: usize, metadata: &Json, + mark_name: &str, ) { let outcome = if call.is_success { "ok" } else { "error" }; let latency_ms = call.duration.as_secs_f64() * 1_000.0; let token_metrics = token_usage_metrics("routing", &call, metadata); events.push(RoutingEvent::Mark(RoutingMark { - name: "switchyard.routing.llm_call".into(), + name: mark_name.into(), data: json!({ "call_index": call_index, "selected_model": call.selected_model.as_str(), diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 0f1738936..3381710f1 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -445,7 +445,7 @@ fn stats_observer( stats.record_error(&call.selected_model); } } - RunObservation::LlmCall(call) => { + RunObservation::LlmCall(call) | RunObservation::DecisionCall(call) => { let latency_ms = call.duration.as_secs_f64() * 1_000.0; if call.is_success { if let (Some((log, context)), Some(usage)) =