From d693aa593dfe9db9442a7191138a8999b79102b9 Mon Sep 17 00:00:00 2001 From: Yehudit Kerido Date: Thu, 27 Aug 2026 13:48:49 +0300 Subject: [PATCH] feat(switchyard-route): add MoM routing filter POC Wire NVIDIA Switchyard Capability-mode classification into Praxis as a decision-only weak/strong router, with a local mock demo so reviewers can verify the plumbing without a cluster. Signed-off-by: Yehudit Kerido --- Cargo.lock | 386 +++++++++++- crates/praxis-experimental-filters/Cargo.toml | 20 + crates/praxis-experimental-filters/src/lib.rs | 8 +- .../src/switchyard_route.rs | 595 ++++++++++++++++++ .../src/switchyard_route/config.rs | 253 ++++++++ demos/README.md | 1 + demos/switchyard-route/.gitignore | 4 + demos/switchyard-route/README.md | 72 +++ demos/switchyard-route/praxis.yaml.template | 42 ++ demos/switchyard-route/run-demo.sh | 110 ++++ demos/switchyard-route/upstreams.py | 153 +++++ deny.toml | 2 + docs/switchyard-route.md | 52 ++ rust-toolchain.toml | 2 +- 14 files changed, 1688 insertions(+), 12 deletions(-) create mode 100644 crates/praxis-experimental-filters/src/switchyard_route.rs create mode 100644 crates/praxis-experimental-filters/src/switchyard_route/config.rs create mode 100644 demos/switchyard-route/.gitignore create mode 100644 demos/switchyard-route/README.md create mode 100644 demos/switchyard-route/praxis.yaml.template create mode 100755 demos/switchyard-route/run-demo.sh create mode 100644 demos/switchyard-route/upstreams.py create mode 100644 docs/switchyard-route.md diff --git a/Cargo.lock b/Cargo.lock index 16163b8..2ec2539 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -18,6 +18,7 @@ dependencies = [ "const-random", "getrandom 0.3.4", "once_cell", + "serde", "version_check", "zerocopy", ] @@ -177,6 +178,28 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-trait" version = "0.1.92" @@ -292,6 +315,12 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "borrow-or-share" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc0b364ead1874514c8c2855ab558056ebfeb775653e7ae45ff72f28f8f3166c" + [[package]] name = "brotli" version = "3.5.0" @@ -330,6 +359,12 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "bytecount" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e" + [[package]] name = "byteorder" version = "1.5.0" @@ -782,6 +817,15 @@ dependencies = [ "serde", ] +[[package]] +name = "email_address" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449" +dependencies = [ + "serde", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -848,6 +892,17 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "fancy-regex" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "476de73bddf2ef8490aa4ee8f1cf40b430bf1d56c48c22080e5186952cd580e6" +dependencies = [ + "bit-set", + "regex-automata", + "regex-syntax", +] + [[package]] name = "fastrand" version = "2.5.0" @@ -877,6 +932,17 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "fluent-uri" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc74ac4d8359ae70623506d512209619e5cf8f347124910440dbc221714b328e" +dependencies = [ + "borrow-or-share", + "ref-cast", + "serde", +] + [[package]] name = "flume" version = "0.12.0" @@ -921,6 +987,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fraction" +version = "0.15.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e076045bb43dac435333ed5f04caf35c7463631d0dae2deb2638d94dd0a5b872" +dependencies = [ + "lazy_static", + "num", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -1080,9 +1156,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 5.3.0", "wasip2", + "wasm-bindgen", ] [[package]] @@ -1588,6 +1666,68 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "jsonptr" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72c22d99362c7aac7d77efab6b49c2181674b165d7e6e7c755b3432ded64bf51" +dependencies = [ + "serde", + "serde_json", +] + +[[package]] +name = "jsonschema" +version = "0.49.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ec8a241beed129f06114aa68007e905ca350e7baeb6e17a7631bb7978d91b2" +dependencies = [ + "ahash", + "bytecount", + "data-encoding", + "email_address", + "fancy-regex 0.19.0", + "fraction", + "getrandom 0.3.4", + "idna", + "itoa", + "jsonschema-regex", + "jsonschema-value", + "num-cmp", + "num-traits", + "percent-encoding", + "referencing", + "regex", + "serde", + "serde_json", + "strum 0.28.0", + "unicode-general-category", + "uuid-simd", +] + +[[package]] +name = "jsonschema-regex" +version = "0.49.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91994f45017ed5e66aa8e59b8415f4cb033a6380d7200387b7cf117595fbdf85" +dependencies = [ + "regex-syntax", +] + +[[package]] +name = "jsonschema-value" +version = "0.49.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ec7637f83e510868ae6ed625f7ebfbbde4554ee8ce49854caa5126a8b9b9ecb" +dependencies = [ + "ahash", + "bytecount", + "fraction", + "num-cmp", + "num-traits", + "serde_json", +] + [[package]] name = "kqueue" version = "1.2.1" @@ -1789,6 +1929,12 @@ dependencies = [ "sketches-ddsketch", ] +[[package]] +name = "micromap" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a86d3146ed3995b5913c414f6664344b9617457320782e64f0bb44afd49d74" + [[package]] name = "minimal-lexical" version = "0.2.1" @@ -1882,6 +2028,20 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" +dependencies = [ + "num-bigint", + "num-complex", + "num-integer", + "num-iter", + "num-rational", + "num-traits", +] + [[package]] name = "num-bigint" version = "0.4.8" @@ -1892,6 +2052,21 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-cmp" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63335b2e2c34fae2fb0aa2cecfd9f0832a1e24b3b32ecec612c3426d46dc8aaa" + +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -1907,6 +2082,27 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-iter" +version = "0.1.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c92800bd69a1eac91786bcfe9da64a897eb72911b8dc3095decbd07429e8048b" +dependencies = [ + "num-integer", + "num-traits", +] + +[[package]] +name = "num-rational" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" +dependencies = [ + "num-bigint", + "num-integer", + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1959,6 +2155,19 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.20", +] + [[package]] name = "ouroboros" version = "0.18.5" @@ -1983,6 +2192,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "outref" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" + [[package]] name = "parking" version = "2.2.1" @@ -2169,7 +2384,20 @@ name = "praxis-experimental-filters" version = "0.1.0" dependencies = [ "async-trait", + "bytes", + "futures", + "http", + "praxis-proxy-core", "praxis-proxy-filter", + "quixotic-plecostomus-core", + "serde", + "serde_json", + "switchyard-libsy", + "switchyard-protocol", + "switchyard-translation", + "thiserror 2.0.20", + "tokio", + "tracing", "yaml_serde", ] @@ -2429,7 +2657,7 @@ dependencies = [ "rmp", "rmp-serde", "serde", - "strum", + "strum 0.26.3", "tokio", ] @@ -2476,8 +2704,8 @@ dependencies = [ "serde_yaml", "sfv", "socket2", - "strum", - "strum_macros", + "strum 0.26.3", + "strum_macros 0.26.4", "tokio", "tokio-stream", "tokio-test", @@ -2766,6 +2994,43 @@ dependencies = [ "bitflags", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "referencing" +version = "0.49.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6efa2154ea6f5ce0fdecdd2a8d18f2fa1a39a8fbba91564f555a592e4dce8278" +dependencies = [ + "ahash", + "fluent-uri", + "getrandom 0.3.4", + "hashbrown 0.17.1", + "itoa", + "micromap", + "parking_lot", + "percent-encoding", + "serde_json", +] + [[package]] name = "regex" version = "1.13.1" @@ -3590,7 +3855,16 @@ version = "0.26.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" dependencies = [ - "strum_macros", + "strum_macros 0.26.4", +] + +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros 0.28.0", ] [[package]] @@ -3606,12 +3880,75 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "subtle" version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "switchyard-libsy" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85f12e1ffefa168604044ccf78535a34dcde38eb59653ac27465e44113cd848c" +dependencies = [ + "async-trait", + "futures", + "jsonptr", + "jsonschema", + "opentelemetry", + "parking_lot", + "rand 0.10.2", + "serde", + "serde_json", + "switchyard-protocol", + "thiserror 2.0.20", + "tokio", + "tokio-stream", + "tracing", + "tracing-opentelemetry", +] + +[[package]] +name = "switchyard-protocol" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c8be18d2d2c600ba679651f9e31ec9eb9dff96060a078c39e50e42ccd4e2318" +dependencies = [ + "async-trait", + "futures", + "http", + "serde", + "serde_json", + "thiserror 2.0.20", +] + +[[package]] +name = "switchyard-translation" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e40cc4f9517e2fa6ecf11c5ffaace82d04689d088ed3095631609e46d2a3f940" +dependencies = [ + "async-stream", + "futures", + "serde", + "serde_json", + "switchyard-protocol", + "thiserror 2.0.20", +] + [[package]] name = "syn" version = "1.0.109" @@ -3736,7 +4073,7 @@ dependencies = [ "anyhow", "base64 0.22.1", "bstr", - "fancy-regex", + "fancy-regex 0.17.0", "lazy_static", "regex", "rustc-hash", @@ -3835,6 +4172,7 @@ dependencies = [ "bytes", "libc", "mio", + "parking_lot", "pin-project-lite", "signal-hook-registry", "socket2", @@ -3988,6 +4326,22 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" +dependencies = [ + "js-sys", + "opentelemetry", + "smallvec", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-serde" version = "0.2.0" @@ -4049,6 +4403,12 @@ version = "0.3.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" +[[package]] +name = "unicode-general-category" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b993bddc193ae5bd0d623b49ec06ac3e9312875fdae725a975c51db1cc1677f" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -4129,6 +4489,16 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "uuid-simd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b082222b4f6619906941c17eb2297fff4c2fb96cb60164170522942a200bd8" +dependencies = [ + "outref", + "vsimd", +] + [[package]] name = "valuable" version = "0.1.1" @@ -4147,6 +4517,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vsimd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" + [[package]] name = "wait-timeout" version = "0.2.1" diff --git a/crates/praxis-experimental-filters/Cargo.toml b/crates/praxis-experimental-filters/Cargo.toml index 6e11806..33b1a74 100644 --- a/crates/praxis-experimental-filters/Cargo.toml +++ b/crates/praxis-experimental-filters/Cargo.toml @@ -19,9 +19,29 @@ workspace = true [dependencies] async-trait = "0.1.89" +bytes = "1.12.1" +futures = "0.3.31" +http = "1.5.0" +serde = { workspace = true } +serde_json = "1.0.151" serde_yaml = { package = "yaml_serde", version = "0.10.4" } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["net"] } +tracing = { workspace = true } # Praxis core (crates.io). Keep in lockstep with praxis-proxy/ai's pins so the # `FilterRegistry` type unifies to a single version across this workspace and the # git-pinned praxis-ai-proxy server crate. praxis-filter = { version = "0.5.3", package = "praxis-proxy-filter" } +praxis-core = { version = "0.5.3", package = "praxis-proxy-core" } + +# The pingora fork praxis builds on; needed for HttpPeer (judge callout targets). +pingora-core = { version = "0.8.2", package = "quixotic-plecostomus-core", features = ["rustls"] } + +# NVIDIA NeMo Switchyard (pre-alpha). Pinned exactly to v0.2.0. +switchyard-libsy = "=0.2.0" +switchyard-protocol = "=0.2.0" +switchyard-translation = "=0.2.0" + +[dev-dependencies] +tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/crates/praxis-experimental-filters/src/lib.rs b/crates/praxis-experimental-filters/src/lib.rs index 70fa4c4..76815df 100644 --- a/crates/praxis-experimental-filters/src/lib.rs +++ b/crates/praxis-experimental-filters/src/lib.rs @@ -5,15 +5,11 @@ //! `server/build.rs` and `praxis-ai-build-support`). The generated registration //! code calls [`register_filters`], which is emitted by the //! [`praxis_filter::export_filters!`] macro invoked below. -//! -//! For now this crate ships a single no-op placeholder filter -//! (`experimental_placeholder`), so that end-to-end discovery and registration -//! are provable before the real filters land (`switchyard_route` in -//! praxis-proxy/experimental#2; `api_key_auth` and `token_ceiling` under the -//! Standalone AI Gateway MVP epic, praxis-proxy/ai#758). mod placeholder; +mod switchyard_route; praxis_filter::export_filters! { http "experimental_placeholder" => placeholder::PlaceholderFilter::from_config, + http "switchyard_route" => switchyard_route::SwitchyardRouteFilter::from_config, } diff --git a/crates/praxis-experimental-filters/src/switchyard_route.rs b/crates/praxis-experimental-filters/src/switchyard_route.rs new file mode 100644 index 0000000..e03d00f --- /dev/null +++ b/crates/praxis-experimental-filters/src/switchyard_route.rs @@ -0,0 +1,595 @@ +//! `switchyard_route`: Mixture-of-Models routing via NVIDIA `NeMo` Switchyard +//! (Capability mode). Decision-only: judge → weak/strong tier → cluster+model. +//! +//! Demo: `judge verdict` / `routed` debug lines are grepped by +//! `demos/switchyard-route/run-demo.sh`. + +#![expect( + clippy::large_futures, + clippy::large_stack_frames, + clippy::too_many_lines, + reason = "POC filter: pingora/switchyard types are large; sequential HTTP logic is clearer inline" +)] + +mod config; + +use std::{sync::Arc, time::Duration}; + +use async_trait::async_trait; +use bytes::Bytes; +use pingora_core::upstreams::peer::HttpPeer; +use praxis_core::subrequest::{SubRequest, SubRequestClient}; +use praxis_filter::{BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, Rejection}; +use switchyard_libsy::Algorithm; +use tracing::{debug, warn}; + +use self::config::{FailureMode, RouteConfig, Tier}; + +/// Metadata key for the chosen cluster (body phase → `on_request`). +const METADATA_CLUSTER: &str = "switchyard_route.cluster"; + +/// Default max body size for buffering (1 MiB). +const DEFAULT_MAX_BODY_BYTES: usize = 1024 * 1024; + +/// The `switchyard_route` HTTP filter. +pub(crate) struct SwitchyardRouteFilter { + /// Validated configuration. + config: RouteConfig, + /// The Capability-mode classifier, built once at config time. + algorithm: Arc, +} + +impl std::fmt::Debug for SwitchyardRouteFilter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SwitchyardRouteFilter") + .field("config", &self.config) + .finish_non_exhaustive() + } +} + +impl SwitchyardRouteFilter { + /// Creates the filter from parsed YAML config. + /// + /// # Errors + /// + /// Returns a [`FilterError`] when the YAML is invalid or Switchyard + /// rejects the classifier configuration. + pub(crate) fn from_config(yaml: &serde_yaml::Value) -> Result, FilterError> { + let config = config::parse(yaml)?; + let algorithm = build_algorithm(&config)?; + Ok(Box::new(Self { config, algorithm })) + } + + /// Runs the routing decision: parse body, call judge, pick tier, rewrite. + async fn route(&self, ctx: &mut HttpFilterContext<'_>, body: &mut Option) -> Result { + let value = parse_body(body.as_ref())?; + + // Detect wire format from path + let path = ctx.request.uri.path(); + if !path.ends_with("/chat/completions") { + return Err(RouteError::UnsupportedPath); + } + + // Decode for judge + let llm_request = decode_for_judge(&value)?; + + // Get subrequest client for judge callout + let client = ctx + .subrequest_client + .as_ref() + .ok_or(RouteError::MissingSubrequestClient)?; + + // Run Switchyard decision loop + let tier = self.decide(client, llm_request).await?; + + // Rewrite model in body + let target = self.config.target(tier); + let new_body = rewrite_model(value, &target.model)?; + *body = Some(Bytes::from(new_body)); + + // Stash cluster for on_request + ctx.set_metadata(METADATA_CLUSTER, target.cluster.clone()); + + Ok(tier) + } + + /// Drives Switchyard's step stream to get a routing decision. + async fn decide( + &self, + client: &SubRequestClient, + llm_request: switchyard_protocol::LlmRequest, + ) -> Result { + use futures::StreamExt as _; + use switchyard_libsy::Step; + use switchyard_protocol::{Context, Metadata, Request}; + + let request = Request { + llm_request, + raw_request: None, + metadata: Some(Metadata::default()), + }; + + let stream = Arc::clone(&self.algorithm).run_stream(Context::default(), request, None); + futures::pin_mut!(stream); + + while let Some(item) = stream.next().await { + let step = item.map_err(|err| RouteError::Run(err.to_string()))?; + match step { + Step::Decision(decision) if decision.is_routed_call() => { + let tag = decision.selected_model(); + return Tier::from_tag(tag).ok_or_else(|| RouteError::UnknownTier(tag.into())); + }, + Step::Decision(_) => {}, + Step::CallLlm(call) => { + if call.get_decision().is_routed_call() { + let tag = call.get_decision().selected_model(); + return Tier::from_tag(tag).ok_or_else(|| RouteError::UnknownTier(tag.into())); + } + // Serve the judge call - unbox the CallLlmRequest + self.serve_judge(client, *call).await?; + }, + Step::ReturnToAgent(_) => return Err(RouteError::NoDecision), + } + } + Err(RouteError::NoDecision) + } + + /// Serves a judge `CallLlm` step via `SubRequestClient`. + /// + /// Switchyard prepares the judge request (system prompt, messages, + /// response format). This method only encodes it onto the `OpenAI` chat + /// wire, POSTs it, and decodes the reply back into Switchyard IR. + async fn serve_judge( + &self, + client: &SubRequestClient, + call: switchyard_libsy::CallLlmRequest, + ) -> Result<(), RouteError> { + let body_bytes = encode_judge_request(&call, &self.config.judge.model)?; + // Demo: useful when debugging judge callouts from `run-demo.sh` / server.log. + debug!(bytes = body_bytes.len(), "switchyard_route: judge request encoded"); + + let endpoint = JudgeEndpoint::parse(&self.config.judge.endpoint)?; + let addrs = resolve_judge_addrs(&endpoint.host, endpoint.port).await?; + let subrequest = endpoint.build_request(body_bytes, self.config.judge.auth_token.as_deref())?; + let timeout = Duration::from_millis(self.config.judge.timeout_ms); + + let callout = JudgeCallout { + client, + endpoint: &endpoint, + subrequest: &subrequest, + timeout, + verify_tls: self.config.judge.verify_tls, + }; + + let mut last_error = String::from("no address attempted"); + for addr in addrs { + match fetch_judge_body(&callout, addr).await { + Ok(body) => { + // Demo: `run-demo.sh` greps this line under "routing decisions". + log_judge_verdict(&body); + match decode_judge_aggregated(addr, &body) { + Ok(aggregated) => return respond_judge(call, aggregated), + Err(JudgeAttemptError::Retryable(message)) => { + last_error = message; + warn!(%last_error, "switchyard_route: judge attempt failed, trying next address"); + }, + Err(JudgeAttemptError::Fatal(err)) => return Err(err), + } + }, + Err(JudgeAttemptError::Retryable(message)) => { + last_error = message; + warn!(%last_error, "switchyard_route: judge attempt failed, trying next address"); + }, + Err(JudgeAttemptError::Fatal(err)) => return Err(err), + } + } + + Err(RouteError::Judge(last_error)) + } +} + +/// Bundles inputs for a judge HTTP callout (keeps argument count down). +struct JudgeCallout<'callout> { + /// Subrequest client from the filter context. + client: &'callout SubRequestClient, + /// Parsed judge URL. + endpoint: &'callout JudgeEndpoint, + /// Encoded OpenAI-style judge POST. + subrequest: &'callout SubRequest, + /// Per-attempt timeout. + timeout: Duration, + /// Whether to verify TLS certificates for HTTPS judges. + verify_tls: bool, +} + +/// Outcome of a single judge address attempt. +enum JudgeAttemptError { + /// Try the next resolved address. + Retryable(String), + /// Stop the callout (translation or respond failure). + Fatal(RouteError), +} + +/// Builds an `HttpPeer` for a judge address, optionally skipping TLS verify. +fn build_judge_peer(addr: std::net::SocketAddr, endpoint: &JudgeEndpoint, verify_tls: bool) -> HttpPeer { + let mut peer = HttpPeer::new(addr, endpoint.tls, endpoint.sni.clone()); + if endpoint.tls && !verify_tls { + peer.options.verify_cert = false; + peer.options.verify_hostname = false; + } + peer +} + +/// POSTs the judge request to one address and returns the response body on 2xx. +async fn fetch_judge_body(callout: &JudgeCallout<'_>, addr: std::net::SocketAddr) -> Result { + let peer = build_judge_peer(addr, callout.endpoint, callout.verify_tls); + let response = callout + .client + .execute(&peer, callout.subrequest, DEFAULT_MAX_BODY_BYTES, callout.timeout, None) + .await + .map_err(|err| JudgeAttemptError::Retryable(format!("{addr}: {err}")))?; + + if !(200..300).contains(&response.status) { + return Err(JudgeAttemptError::Retryable(format!( + "HTTP {} from {addr} body_len={} preview={:?}", + response.status, + response.body.len(), + body_preview(&response.body) + ))); + } + Ok(response.body) +} + +/// Parses and translates a judge JSON body into Switchyard IR. +fn decode_judge_aggregated( + addr: std::net::SocketAddr, + body: &Bytes, +) -> Result { + use switchyard_protocol::WireFormat; + + let value: serde_json::Value = serde_json::from_slice(body).map_err(|err| { + JudgeAttemptError::Retryable(format!( + "judge_non_json from {addr}: {err}; body_len={} preview={:?}", + body.len(), + body_preview(body) + )) + })?; + + switchyard_translation::decode_aggregated_response(&value, WireFormat::OpenAiChat) + .map_err(|err| JudgeAttemptError::Fatal(RouteError::Judge(format!("response translation failed: {err}")))) +} + +/// Delivers a decoded judge response back into the Switchyard step loop. +fn respond_judge( + call: switchyard_libsy::CallLlmRequest, + aggregated: switchyard_protocol::AggLlmResponse, +) -> Result<(), RouteError> { + use switchyard_protocol::{LlmResponse, Response}; + + call.respond(Ok(Response { + llm_response: LlmResponse::Agg(aggregated), + metadata: None, + })) + .map_err(|err| RouteError::Judge(format!("failed to deliver judge response: {err}"))) +} + +#[async_trait] +impl HttpFilter for SwitchyardRouteFilter { + fn name(&self) -> &'static str { + "switchyard_route" + } + + fn request_body_access(&self) -> BodyAccess { + BodyAccess::ReadWrite + } + + fn request_body_mode(&self) -> BodyMode { + BodyMode::StreamBuffer { + max_bytes: Some(DEFAULT_MAX_BODY_BYTES), + } + } + + fn selects_cluster(&self) -> bool { + true + } + + fn selected_clusters(&self) -> Vec { + vec![self.config.weak.cluster.clone(), self.config.strong.cluster.clone()] + } + + async fn on_request_body( + &self, + ctx: &mut HttpFilterContext<'_>, + body: &mut Option, + end_of_stream: bool, + ) -> Result { + // StreamBuffer may invoke this before the full body is available. + if !end_of_stream { + return Ok(FilterAction::Continue); + } + + match self.route(ctx, body).await { + Ok(tier) => { + // Demo: `run-demo.sh` greps `switchyard_route: routed`. + debug!(tier = %tier.tag(), "switchyard_route: routed"); + Ok(FilterAction::Continue) + }, + Err(err) => { + // Demo: `run-demo.sh` greps `switchyard_route: routing failed`. + warn!(error = %err, "switchyard_route: routing failed"); + let short = err.to_string(); + let short: String = short.chars().take(250).collect(); + ctx.set_metadata("switchyard_route.error", short); + + match self.config.on_failure { + FailureMode::Open => { + // Demo: `run-demo.sh` greps `switchyard_route: fail-open`. + debug!("switchyard_route: fail-open, passing through"); + Ok(FilterAction::Continue) + }, + FailureMode::Closed => Ok(FilterAction::Reject(Rejection::status(503))), + } + }, + } + } + + async fn on_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result { + if let Some(cluster) = ctx.get_metadata(METADATA_CLUSTER) { + ctx.cluster = Some(cluster.into()); + } + Ok(FilterAction::Continue) + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Parsed judge URL components for the sub-request callout. +struct JudgeEndpoint { + /// Whether the judge endpoint uses HTTPS. + tls: bool, + /// Hostname used for DNS resolution. + host: String, + /// TCP port. + port: u16, + /// TLS SNI; empty for cleartext HTTP. + sni: String, + /// Original URL authority, sent as the `Host` header. + authority: http::HeaderValue, + /// Path and query for the POST. + uri: http::Uri, +} + +impl JudgeEndpoint { + /// Parses an absolute http(s) judge URL. + fn parse(endpoint: &str) -> Result { + let parsed: http::Uri = endpoint + .parse() + .map_err(|err| RouteError::Judge(format!("bad endpoint: {err}")))?; + let tls = match parsed.scheme_str() { + Some("https") => true, + Some("http") => false, + Some(other) => return Err(RouteError::Judge(format!("unsupported scheme '{other}'"))), + None => return Err(RouteError::Judge("endpoint must be an absolute http(s) URL".into())), + }; + let authority = parsed + .authority() + .ok_or_else(|| RouteError::Judge("endpoint missing host".into()))?; + let host = authority + .host() + .trim_start_matches('[') + .trim_end_matches(']') + .to_owned(); + let port = authority.port_u16().unwrap_or(if tls { 443 } else { 80 }); + let sni = if tls { host.clone() } else { String::new() }; + let authority = http::HeaderValue::from_str(authority.as_str()) + .map_err(|err| RouteError::Judge(format!("invalid authority: {err}")))?; + let uri: http::Uri = parsed + .path_and_query() + .map_or("/", http::uri::PathAndQuery::as_str) + .parse() + .map_err(|err| RouteError::Judge(format!("bad path: {err}")))?; + Ok(Self { + tls, + host, + port, + sni, + authority, + uri, + }) + } + + /// Builds the POST sub-request carrying the encoded judge body. + fn build_request(&self, body: Bytes, auth_token: Option<&str>) -> Result { + let mut headers = http::HeaderMap::new(); + headers.insert(http::header::HOST, self.authority.clone()); + headers.insert( + http::header::CONTENT_TYPE, + http::HeaderValue::from_static("application/json"), + ); + if let Some(token) = auth_token { + let value = format!("Bearer {token}"); + let header = http::HeaderValue::from_str(&value) + .map_err(|err| RouteError::Judge(format!("invalid auth header: {err}")))?; + headers.insert(http::header::AUTHORIZATION, header); + } + Ok(SubRequest { + method: http::Method::POST, + uri: self.uri.clone(), + headers, + body, + }) + } +} + +/// Resolves every address for the judge host so connects can fall back. +async fn resolve_judge_addrs(host: &str, port: u16) -> Result, RouteError> { + let addrs: Vec = tokio::net::lookup_host((host, port)) + .await + .map_err(|err| RouteError::Judge(format!("DNS resolution failed for {host}: {err}")))? + .collect(); + if addrs.is_empty() { + return Err(RouteError::Judge(format!("no addresses resolved for {host}"))); + } + Ok(addrs) +} + +/// Truncates a body for log-safe error previews. +fn body_preview(body: &Bytes) -> String { + String::from_utf8_lossy(body).chars().take(200).collect() +} + +/// Builds the Capability-mode classifier from config. +fn build_algorithm(config: &RouteConfig) -> Result, FilterError> { + use switchyard_libsy::{ + ClassifierContractConfig, LlmClassifierConfig, LlmTarget, LlmTaskClassifier, TaskClassifierConfig, + }; + + let classifier_config = LlmClassifierConfig::Capability { + judge_target: LlmTarget { + semantic_name: "judge".to_owned(), + llm_client: None, + }, + efficient_target: LlmTarget { + semantic_name: Tier::Weak.tag().to_owned(), + llm_client: None, + }, + capable_target: LlmTarget { + semantic_name: Tier::Strong.tag().to_owned(), + llm_client: None, + }, + config: TaskClassifierConfig { + base_threshold: config.threshold, + session_affinity: false, + contract: ClassifierContractConfig::default(), + ..TaskClassifierConfig::default() + }, + }; + + let classifier = LlmTaskClassifier::new(classifier_config) + .map_err(|err| FilterError::from(format!("switchyard config rejected: {err}")))?; + + let arc: Arc = Arc::new(classifier); + Ok(arc) +} + +/// Encodes the prepared judge request onto the `OpenAI` chat wire. +fn encode_judge_request(call: &switchyard_libsy::CallLlmRequest, judge_model: &str) -> Result { + let mut llm_request = call.get_request().llm_request.clone(); + llm_request.model = Some(judge_model.to_owned()); + llm_request.stream = false; + let wire = switchyard_translation::encode_request(&llm_request, switchyard_protocol::WireFormat::OpenAiChat) + .map_err(|err| RouteError::Judge(format!("request encoding failed: {err}")))?; + let bytes = + serde_json::to_vec(&wire).map_err(|err| RouteError::Judge(format!("request serialization failed: {err}")))?; + Ok(Bytes::from(bytes)) +} + +/// Demo: log Switchyard verdict fields for `run-demo.sh` ("routing decisions"). +fn log_judge_verdict(body: &[u8]) { + let Ok(value) = serde_json::from_slice::(body) else { + return; + }; + let Some(content) = value + .pointer("/choices/0/message/content") + .and_then(serde_json::Value::as_str) + else { + return; + }; + let Ok(verdict) = serde_json::from_str::(content) else { + return; + }; + let p_solve = verdict.get("p_solve").and_then(serde_json::Value::as_f64); + let rule = verdict + .get("primary_rule") + .and_then(serde_json::Value::as_str) + .unwrap_or("?"); + let boundary = verdict + .get("capability_boundary") + .and_then(serde_json::Value::as_str) + .unwrap_or("?"); + debug!(?p_solve, rule, boundary, "switchyard_route: judge verdict"); +} + +/// Parses the buffered body as JSON. +fn parse_body(body: Option<&Bytes>) -> Result { + let raw = body.ok_or(RouteError::Body("missing"))?; + if raw.is_empty() { + return Err(RouteError::Body("empty")); + } + serde_json::from_slice(raw).map_err(|err| RouteError::Json(err.to_string())) +} + +/// Decodes an `OpenAI` chat body into Switchyard IR for the judge. +fn decode_for_judge(body: &serde_json::Value) -> Result { + switchyard_translation::decode_request(switchyard_protocol::WireFormat::OpenAiChat, body) + .map_err(|err| RouteError::Translation(err.to_string())) +} + +/// Rewrites the `model` field and re-serializes. +fn rewrite_model(mut body: serde_json::Value, model: &str) -> Result, RouteError> { + body.as_object_mut() + .ok_or(RouteError::Body("not an object"))? + .insert("model".to_owned(), serde_json::Value::String(model.into())); + serde_json::to_vec(&body).map_err(|err| RouteError::Serialize(err.to_string())) +} + +// --------------------------------------------------------------------------- +// Errors +// --------------------------------------------------------------------------- + +/// Routing failures (internal to this filter). +#[derive(Debug, thiserror::Error)] +enum RouteError { + /// Request body missing, empty, or not a JSON object. + #[error("request body {0}")] + Body(&'static str), + /// Body bytes were not valid JSON. + #[error("invalid JSON: {0}")] + Json(String), + /// Path is not an `OpenAI` chat completions endpoint. + #[error("unsupported path (only /chat/completions)")] + UnsupportedPath, + /// `OpenAI` ↔ Switchyard IR translation failed. + #[error("translation failed: {0}")] + Translation(String), + /// Failed to re-serialize the rewritten request body. + #[error("serialize failed: {0}")] + Serialize(String), + /// Filter context had no `SubRequestClient` for the judge callout. + #[error("subrequest client unavailable")] + MissingSubrequestClient, + /// Judge HTTP callout failed after retries. + #[error("judge callout failed: {0}")] + Judge(String), + /// Switchyard `run_stream` returned an error. + #[error("switchyard run failed: {0}")] + Run(String), + /// Stream ended without a routed weak/strong decision. + #[error("no routing decision")] + NoDecision, + /// Decision tag was not `weak` or `strong`. + #[error("unknown tier '{0}'")] + UnknownTier(String), +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use praxis_filter::FilterRegistry; + + #[test] + fn filter_is_registered() { + let mut registry = FilterRegistry::with_builtins(); + crate::register_filters(&mut registry); + let names = registry.available_filters(); + assert!( + names.contains(&"switchyard_route"), + "expected switchyard_route in {names:?}" + ); + } +} diff --git a/crates/praxis-experimental-filters/src/switchyard_route/config.rs b/crates/praxis-experimental-filters/src/switchyard_route/config.rs new file mode 100644 index 0000000..285ab3a --- /dev/null +++ b/crates/praxis-experimental-filters/src/switchyard_route/config.rs @@ -0,0 +1,253 @@ +//! Configuration for the `switchyard_route` filter. + +// Under `cfg(test)` this lint may not fire, which would leave a module-level +// `expect` unfulfilled; only suppress when compiling the library normally. +#![cfg_attr( + not(test), + expect( + clippy::missing_docs_in_private_items, + reason = "config types are straightforward data containers" + ) +)] + +use praxis_filter::FilterError; +use serde::Deserialize; + +/// Validated filter configuration. +#[derive(Debug, Clone)] +pub(crate) struct RouteConfig { + pub(crate) judge: JudgeConfig, + pub(crate) weak: TargetConfig, + pub(crate) strong: TargetConfig, + pub(crate) threshold: f64, + pub(crate) on_failure: FailureMode, +} + +impl RouteConfig { + pub(crate) fn target(&self, tier: Tier) -> &TargetConfig { + match tier { + Tier::Weak => &self.weak, + Tier::Strong => &self.strong, + } + } +} + +/// Judge (classifier LLM) callout settings. +#[derive(Clone)] +pub(crate) struct JudgeConfig { + pub(crate) endpoint: String, + pub(crate) model: String, + pub(crate) timeout_ms: u64, + pub(crate) verify_tls: bool, + /// Bearer token from `auth.value_env` at startup, if configured. + pub(crate) auth_token: Option, +} + +impl std::fmt::Debug for JudgeConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("JudgeConfig") + .field("endpoint", &self.endpoint) + .field("model", &self.model) + .field("timeout_ms", &self.timeout_ms) + .field("verify_tls", &self.verify_tls) + .field("auth_token", &self.auth_token.as_ref().map(|_| "[redacted]")) + .finish() + } +} + +/// Target cluster + model for a tier. +#[derive(Debug, Clone)] +pub(crate) struct TargetConfig { + pub(crate) cluster: String, + pub(crate) model: String, +} + +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(rename_all = "lowercase")] +pub(crate) enum FailureMode { + /// Pass through unchanged on failure. + #[default] + Open, + /// Reject with 503 on failure. + Closed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) enum Tier { + Weak, + Strong, +} + +impl Tier { + pub(crate) fn tag(self) -> &'static str { + match self { + Self::Weak => "weak", + Self::Strong => "strong", + } + } + + pub(crate) fn from_tag(tag: &str) -> Option { + match tag { + "weak" => Some(Self::Weak), + "strong" => Some(Self::Strong), + _ => None, + } + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawConfig { + judge: RawJudge, + targets: RawTargets, + #[serde(default = "default_threshold")] + threshold: f64, + #[serde(default)] + on_failure: FailureMode, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawJudge { + endpoint: String, + model: String, + #[serde(default = "default_timeout")] + timeout_ms: u64, + #[serde(default = "default_verify_tls")] + verify_tls: bool, + #[serde(default)] + auth: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawJudgeAuth { + value_env: String, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawTargets { + weak: RawTarget, + strong: RawTarget, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawTarget { + cluster: String, + model: String, +} + +fn default_threshold() -> f64 { + 0.5 +} + +fn default_timeout() -> u64 { + 5000 +} + +fn default_verify_tls() -> bool { + true +} + +/// Parses and validates YAML config. +pub(crate) fn parse(yaml: &serde_yaml::Value) -> Result { + let raw: RawConfig = serde_yaml::from_value(yaml.clone()).map_err(|err| FilterError::from(err.to_string()))?; + + if !(0.0..=1.0).contains(&raw.threshold) { + return Err(FilterError::from("threshold must be between 0.0 and 1.0")); + } + if raw.judge.endpoint.parse::().is_err() { + return Err(FilterError::from("judge.endpoint is not a valid URL")); + } + + let auth_token = match raw.judge.auth { + None => None, + Some(auth) => { + if auth.value_env.is_empty() { + return Err(FilterError::from("judge.auth.value_env must not be empty")); + } + let value = std::env::var(&auth.value_env).map_err(|_err| { + FilterError::from(format!("judge.auth.value_env '{}' is unset or empty", auth.value_env)) + })?; + if value.is_empty() { + return Err(FilterError::from(format!( + "judge.auth.value_env '{}' is unset or empty", + auth.value_env + ))); + } + Some(value) + }, + }; + + Ok(RouteConfig { + judge: JudgeConfig { + endpoint: raw.judge.endpoint, + model: raw.judge.model, + timeout_ms: raw.judge.timeout_ms, + verify_tls: raw.judge.verify_tls, + auth_token, + }, + weak: TargetConfig { + cluster: raw.targets.weak.cluster, + model: raw.targets.weak.model, + }, + strong: TargetConfig { + cluster: raw.targets.strong.cluster, + model: raw.targets.strong.model, + }, + threshold: raw.threshold, + on_failure: raw.on_failure, + }) +} + +#[cfg(test)] +#[expect( + clippy::unwrap_used, + clippy::expect_used, + clippy::indexing_slicing, + clippy::assertions_on_result_states, + reason = "tests" +)] +mod tests { + use super::*; + + fn valid_yaml() -> serde_yaml::Value { + serde_yaml::from_str( + r#" +judge: + endpoint: "http://localhost:8000/v1/chat/completions" + model: "judge-model" +targets: + weak: + cluster: "weak-cluster" + model: "weak-model" + strong: + cluster: "strong-cluster" + model: "strong-model" +"#, + ) + .unwrap() + } + + #[test] + fn parses_minimal_config() { + let config = parse(&valid_yaml()).expect("should parse"); + assert_eq!(config.judge.model, "judge-model"); + assert_eq!(config.weak.cluster, "weak-cluster"); + assert!((config.threshold - 0.5).abs() < f64::EPSILON); + } + + #[test] + fn rejects_invalid_threshold() { + let mut yaml = valid_yaml(); + yaml["threshold"] = serde_yaml::Value::Number(serde_yaml::Number::from(1.5)); + assert!(parse(&yaml).is_err()); + } + + #[test] + fn tier_ordering() { + assert!(Tier::Weak < Tier::Strong); + } +} diff --git a/demos/README.md b/demos/README.md index da1cf92..d7458a6 100644 --- a/demos/README.md +++ b/demos/README.md @@ -6,3 +6,4 @@ Demonstrations of experimental Praxis features live in this directory. - [Praxis Grid - Distributed token rate limiting with Grid routing](grid-distributed-token-rate-limit/README.md) - [Praxis Grid - Intelligent Overflow](grid-cloud-burst/README.md) +- [Mixture-of-Models routing with Switchyard](switchyard-route/README.md) diff --git a/demos/switchyard-route/.gitignore b/demos/switchyard-route/.gitignore new file mode 100644 index 0000000..fe6ea98 --- /dev/null +++ b/demos/switchyard-route/.gitignore @@ -0,0 +1,4 @@ +# Generated by run-demo.sh +praxis.yaml +server.log +.auth-block.yaml diff --git a/demos/switchyard-route/README.md b/demos/switchyard-route/README.md new file mode 100644 index 0000000..ca19c3f --- /dev/null +++ b/demos/switchyard-route/README.md @@ -0,0 +1,72 @@ +# `switchyard_route` demo + +Local end-to-end POC for the [`switchyard_route`](../../docs/switchyard-route.md) +filter: a mock Switchyard judge classifies each chat request as easy or hard, +then Praxis routes to a weak or strong echo upstream. + +No Kubernetes cluster and no real LLM — only loopback mocks. + +## What you should see + +1. Three easy prompts → `served_by=weak-upstream` +2. Three hard prompts → `served_by=strong-upstream` +3. Gateway logs with `switchyard_route: judge verdict` / `routed` +4. Mock logs showing judge `p_solve` and which upstream answered + +## Quick start + +```console +cd demos/switchyard-route +./run-demo.sh +``` + +The script: + +1. Starts `upstreams.py` (judge `:18091`, weak `:18092`, strong `:18093`) +2. Builds `praxis-experimental-server` if needed +3. Renders `praxis.yaml` from `praxis.yaml.template` +4. Starts the gateway on `:18080` +5. Sends 3 easy + 3 hard prompts and greps the logs + +## Ports + +| Role | Port | Behavior | +| --- | --- | --- | +| Gateway | `:18080` | Praxis + `switchyard_route` + `load_balancer` | +| Judge | `:18091` | Easy → `p_solve=0.95` / `SUP-1`; hard markers → `0.0` / `LIM-2` | +| Weak upstream | `:18092` | Echo `served_by=weak-upstream` | +| Strong upstream | `:18093` | Echo `served_by=strong-upstream` | + +Hard prompts include markers such as `undocumented`, `blurry`, `whiteboard` +(see `_HARD_MARKERS` in `upstreams.py`). With `threshold: 0.8` in the demo +YAML, `0.95` routes weak and `0.0` routes strong. + +## Files + +| File | Role | +| --- | --- | +| `run-demo.sh` | One-shot demo driver | +| `upstreams.py` | Mock judge + weak/strong echo servers | +| `praxis.yaml.template` | Full Praxis config (placeholders for judge) | +| `praxis.yaml` | Generated at run time (gitignored) | +| `server.log` | Symlink to gateway log (gitignored) | + +## Mocks only + +To run the three mock servers without Praxis: + +```console +python3 upstreams.py +``` + +## Layout (request path) + +```text +Client + → Gateway :18080 + → switchyard_route (judge callout :18091) + → load_balancer + → weak :18092 or strong :18093 +``` + +Filter docs: [`docs/switchyard-route.md`](../../docs/switchyard-route.md). diff --git a/demos/switchyard-route/praxis.yaml.template b/demos/switchyard-route/praxis.yaml.template new file mode 100644 index 0000000..48946ec --- /dev/null +++ b/demos/switchyard-route/praxis.yaml.template @@ -0,0 +1,42 @@ +# switchyard_route local demo (mocks on 127.0.0.1). +# +# gateway :18080 → switchyard_route → judge :18091 / weak :18092 / strong :18093 + +admin: + address: "127.0.0.1:19901" + +listeners: + - name: switchyard-gateway + address: "127.0.0.1:18080" + filter_chains: + - routing + +filter_chains: + - name: routing + filters: + - filter: switchyard_route + judge: + endpoint: "__JUDGE_ENDPOINT__" + model: "__JUDGE_MODEL__" + timeout_ms: 10000 + threshold: 0.8 + targets: + weak: + cluster: weak-cluster + model: mock-weak + strong: + cluster: strong-cluster + model: mock-strong + on_failure: open + - filter: load_balancer + clusters: + - name: weak-cluster + endpoints: + - "127.0.0.1:18092" + - name: strong-cluster + endpoints: + - "127.0.0.1:18093" + +# Loopback-only demo; allow_root for UID 0 CI/container shells. +insecure_options: + allow_root: true diff --git a/demos/switchyard-route/run-demo.sh b/demos/switchyard-route/run-demo.sh new file mode 100755 index 0000000..184629d --- /dev/null +++ b/demos/switchyard-route/run-demo.sh @@ -0,0 +1,110 @@ +#!/usr/bin/env bash +set -euo pipefail + +# switchyard_route local mock demo. +# Optional: JUDGE_ENDPOINT, JUDGE_MODEL, FORCE_REBUILD=1 + +cd "$(dirname "$0")" + +GATEWAY=http://127.0.0.1:18080/v1/chat/completions +REPO_ROOT="$(cd ../.. && pwd)" +export CARGO_TARGET_DIR="${REPO_ROOT}/target" +SERVER_BIN="${CARGO_TARGET_DIR}/debug/praxis-experimental-server" +JUDGE_ENDPOINT="${JUDGE_ENDPOINT:-http://127.0.0.1:18091/v1/chat/completions}" +JUDGE_MODEL="${JUDGE_MODEL:-mock-switchyard-judge}" +MOCKS_PID="" + +cleanup() { + kill "${SERVER_PID:-}" 2>/dev/null || true + kill "${MOCKS_PID:-}" 2>/dev/null || true +} +trap cleanup EXIT + +echo "mode: local mocks (judge :18091, weak :18092, strong :18093)" >&2 +python3 upstreams.py > /tmp/switchyard-demo-mocks.log 2>&1 & +MOCKS_PID=$! +sleep 0.3 +if ! kill -0 "$MOCKS_PID" 2>/dev/null; then + echo "mock servers failed; see /tmp/switchyard-demo-mocks.log" >&2 + exit 1 +fi + +FILTER_SRC="${REPO_ROOT}/crates/praxis-experimental-filters/src/switchyard_route.rs" +if [[ "${FORCE_REBUILD:-}" == "1" ]] \ + || [[ ! -x "$SERVER_BIN" ]] \ + || [[ "$FILTER_SRC" -nt "$SERVER_BIN" ]]; then + echo "building praxis-experimental-server..." >&2 + (cd "$REPO_ROOT" && cargo build -p praxis-experimental-server) +fi + +sed -e "s|__JUDGE_ENDPOINT__|${JUDGE_ENDPOINT}|" \ + -e "s|__JUDGE_MODEL__|${JUDGE_MODEL}|" \ + -e "/__JUDGE_AUTH_BLOCK__/d" \ + praxis.yaml.template > praxis.yaml + +pkill -f 'praxis-experimental-server' 2>/dev/null || true +sleep 0.2 + +RUST_LOG="${RUST_LOG:-info,praxis_experimental_filters=debug}" \ + "$SERVER_BIN" > /tmp/switchyard-demo-server.log 2>&1 & +SERVER_PID=$! +ln -sfn /tmp/switchyard-demo-server.log server.log + +for _ in $(seq 1 60); do + if curl -sf -o /dev/null -m 10 -X POST "$GATEWAY" \ + -H 'content-type: application/json' \ + -d '{"model":"warmup","messages":[{"role":"user","content":"ping"}],"max_tokens":1}' 2>/dev/null; then + break + fi + if ! kill -0 "$SERVER_PID" 2>/dev/null; then + echo "server exited early; see server.log:" >&2 + tail -n 40 /tmp/switchyard-demo-server.log >&2 || true + exit 1 + fi + sleep 0.5 +done + +ask() { + local label=$1 prompt=$2 tmp body http + tmp=$(mktemp) + body=$(PROMPT="$prompt" python3 - <<'PY' +import json, os +print(json.dumps({ + "model": "agent-default", + "messages": [{"role": "user", "content": os.environ["PROMPT"]}], + "max_tokens": 64, + "stream": False, +})) +PY +) + echo "--- ${label} ---" + echo "prompt: ${prompt}" + http=$(curl -sS -m 60 -o "$tmp" -w '%{http_code}' -X POST "$GATEWAY" \ + -H 'content-type: application/json' -d "$body" || true) + echo "HTTP ${http}" + if [[ -s "$tmp" ]]; then + python3 -c 'import json,sys; r=json.load(open(sys.argv[1])); print(r["choices"][0]["message"]["content"][:200])' "$tmp" 2>/dev/null \ + || cat "$tmp" + echo + else + echo "(empty body — see server.log)" + fi + rm -f "$tmp" +} + +echo "=== easy (expect weak) ===" +ask easy1 'What is 2+2?' +ask easy2 'What is the capital of France?' +ask easy3 'Translate hello into Spanish. One word only.' + +echo "=== hard (expect strong) ===" +ask hard1 'Reverse-engineer an undocumented legacy billing service with no harness.' +ask hard2 'From a blurry whiteboard photo with no image or OCR, recover every equation.' +ask hard3 'Reproduce undocumented acme-vision tensor layouts with no golden files.' + +echo +echo "routing decisions (ignore warmup):" +grep -E 'switchyard_route: (judge verdict|routed|routing failed|fail-open)' /tmp/switchyard-demo-server.log || true +echo +echo "upstreams (4× weak = warmup + 3 easy, then 3× strong):" +grep -nE 'weak-upstream|strong-upstream' /tmp/switchyard-demo-mocks.log || true diff --git a/demos/switchyard-route/upstreams.py b/demos/switchyard-route/upstreams.py new file mode 100644 index 0000000..eeea098 --- /dev/null +++ b/demos/switchyard-route/upstreams.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""Local mocks for switchyard_route: judge :18091, weak :18092, strong :18093.""" + +from __future__ import annotations + +import json +import re +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +JUDGE_PORT, WEAK_PORT, STRONG_PORT = 18091, 18092, 18093 + +# Markers that flip the mock judge to LIM-2 / p_solve=0 (demo hard prompts). +_HARD_MARKERS = ( + "reverse-engineer", + "undocumented", + "blurry", + "whiteboard", + "acme-vision", + "no harness", + "golden file", +) + + +def _read_json(handler: BaseHTTPRequestHandler) -> dict[str, Any]: + length = int(handler.headers.get("Content-Length", "0")) + raw = handler.rfile.read(length) if length else b"{}" + try: + value = json.loads(raw or b"{}") + except json.JSONDecodeError: + return {} + return value if isinstance(value, dict) else {} + + +def _write_json(handler: BaseHTTPRequestHandler, payload: dict[str, Any]) -> None: + body = json.dumps(payload).encode() + handler.send_response(200) + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + handler.wfile.write(body) + + +def _latest_user_text(body: dict[str, Any]) -> str: + messages = body.get("messages") + if not isinstance(messages, list): + return "" + for message in reversed(messages): + if isinstance(message, dict) and message.get("role") == "user": + content = message.get("content") + if isinstance(content, str): + return content + return "" + + +def _verdict(prompt: str) -> dict[str, Any]: + lowered = prompt.lower() + if any(marker in lowered for marker in _HARD_MARKERS): + return { + "crux": "undocumented reference or missing harness", + "primary_rule": "LIM-2", + "capability_boundary": "unsupported", + "p_solve": 0.0, + } + return { + "crux": "bounded factual task", + "primary_rule": "SUP-1", + "capability_boundary": "supported", + "p_solve": 0.95, + } + + +def _chat_completion(model: str, content: str) -> dict[str, Any]: + return { + "id": "chat-mock", + "object": "chat.completion", + "model": model, + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": content}, + "finish_reason": "stop", + } + ], + } + + +def judge_handler() -> type[BaseHTTPRequestHandler]: + class Judge(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + if not self.path.startswith("/v1/chat/completions"): + self.send_error(404) + return + body = _read_json(self) + prompt = _latest_user_text(body) + match = re.search(r"(?is)user:\s*(.+)$", prompt) + if match: + prompt = match.group(1).strip() + verdict = _verdict(prompt) + print( + f"[judge] p_solve={verdict['p_solve']} rule={verdict['primary_rule']} " + f"preview={prompt[:60]!r}", + flush=True, + ) + _write_json( + self, + _chat_completion( + body.get("model") or "mock-switchyard-judge", + json.dumps(verdict, separators=(",", ":")), + ), + ) + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 + return + + return Judge + + +def upstream(name: str) -> type[BaseHTTPRequestHandler]: + class Upstream(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + body = _read_json(self) + _write_json( + self, + _chat_completion(body.get("model") or name, f"served_by={name}"), + ) + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 + print(f"[{name}] {format % args}", flush=True) + + return Upstream + + +def serve(port: int, handler: type[BaseHTTPRequestHandler]) -> ThreadingHTTPServer: + server = ThreadingHTTPServer(("127.0.0.1", port), handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + print(f"listening on 127.0.0.1:{port}", flush=True) + return server + + +if __name__ == "__main__": + servers = [ + serve(JUDGE_PORT, judge_handler()), + serve(WEAK_PORT, upstream("weak-upstream")), + serve(STRONG_PORT, upstream("strong-upstream")), + ] + print("mocks ready", flush=True) + try: + threading.Event().wait() + except KeyboardInterrupt: + for running in servers: + running.shutdown() diff --git a/deny.toml b/deny.toml index 0664f96..fc519f7 100644 --- a/deny.toml +++ b/deny.toml @@ -25,6 +25,8 @@ allow = [ "CDLA-Permissive-2.0", "ISC", "MIT", + # Via Switchyard → jsonschema → referencing → fluent-uri → borrow-or-share. + "MIT-0", "Unicode-3.0", "Unicode-DFS-2016", "Unlicense", diff --git a/docs/switchyard-route.md b/docs/switchyard-route.md new file mode 100644 index 0000000..d44966c --- /dev/null +++ b/docs/switchyard-route.md @@ -0,0 +1,52 @@ +# `switchyard_route`: Capability-mode Mixture-of-Models routing + +> **Status: POC** ([praxis-proxy/experimental#2](https://github.com/praxis-proxy/experimental/issues/2)). +> Built against NVIDIA NeMo Switchyard `=0.2.0` (pre-alpha). + +Decision-only router: a judge classifies each request; Switchyard returns +`weak` / `strong`; the filter maps that tag to `(cluster, model)` and selects +the Praxis cluster. Switchyard never sees provider names. + +## Flow + +1. **`on_request_body`**: buffer JSON, decode OpenAI chat → Switchyard IR, + drive `run_stream`, serve the judge `CallLlm` via `SubRequestClient`, + rewrite `model`, stash cluster metadata. +2. **`on_request`**: apply `ctx.cluster` from metadata. + +Metadata: `switchyard_route.cluster` on success; `switchyard_route.error` on +failure. + +## Configuration + +```yaml +- filter: switchyard_route + judge: + endpoint: "http://127.0.0.1:18091/v1/chat/completions" + model: mock-switchyard-judge + # auth: + # value_env: OPENAI_API_KEY + timeout_ms: 5000 + threshold: 0.8 + targets: + weak: + cluster: weak-cluster + model: mock-weak + strong: + cluster: strong-cluster + model: mock-strong + on_failure: open # open | closed +``` + +- Path: `*/chat/completions` only. +- Secrets: `judge.auth.value_env` only (never inline). +- `on_failure: open` passes through; `closed` → HTTP 503. + +## Demo + +```console +cd demos/switchyard-route && ./run-demo.sh +``` + +Mock judge + echo upstreams. Easy → `served_by=weak-upstream`; hard → +`served_by=strong-upstream`. Details in [`demos/switchyard-route/`](../demos/switchyard-route/README.md). diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 61d8532..a925d26 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,3 +1,3 @@ [toolchain] -channel = "1.96.0" +channel = "1.96.1" components = ["clippy", "rust-analyzer"]